starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-27 00:02:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

v1

Browse Source
This commit is contained in:
Ruben Fiszel
2025-07-14 16:49:00 +00:00
parent fff5455f29
commit 904e07baf3
11 changed files with 132 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/openapi.yaml
+5 -1
View File
@@ -3099,7 +3099,7 @@ paths:
type: array
items:
$ref: "#/components/schemas/ContextualVariable"
/w/{workspace}/workspaces/get_secondary_storage_names:
get:
summary: get secondary storage names
@@ -7488,6 +7488,10 @@ paths:
in: query
schema:
type: boolean
- name: no_code
in: query
schema:
type: boolean
responses:
"200":
description: job details
backend/windmill-api/src/jobs.rs
+42 -28
View File
@@ -527,7 +527,7 @@ async fn get_flow_job_debug_info(
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
let job = GetQuery::new()
.fetch_queued((&db).into(), id, &w_id)
.fetch_queued((&db).into(), &id, &w_id)
.await?;
if let Some(job) = job {
let is_flow = job.is_flow();
@@ -567,7 +567,7 @@ async fn get_flow_job_debug_info(
for job_id in job_ids {
let job = GetQuery::new()
.with_auth(&opt_authed)
.fetch(&db, job_id, &w_id)
.fetch(&db, &job_id, &w_id)
.await;
if let Ok(job) = job {
jobs.insert(job.id().to_string(), job);
@@ -637,6 +637,7 @@ async fn list_selected_job_groups(
#[derive(Deserialize)]
struct GetJobQuery {
pub no_logs: Option<bool>,
pub no_code: Option<bool>,
}
async fn get_job(
@@ -644,7 +645,7 @@ async fn get_job(
opt_tokened: OptTokened,
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
Query(GetJobQuery { no_logs }): Query<GetJobQuery>,
Query(GetJobQuery { no_logs, no_code }): Query<GetJobQuery>,
) -> error::Result<Response> {
let tags = opt_authed
.as_ref()
@@ -655,10 +656,14 @@ async fn get_job(
.with_auth(&opt_authed)
.with_in_tags(tags.as_ref());
if no_code.unwrap_or(false) {
get = get.without_code();
}
if no_logs.unwrap_or(false) {
get = get.without_logs();
}
let mut job = get.fetch(&db, id, &w_id).await?;
let mut job = get.fetch(&db, &id, &w_id).await?;
job.fetch_outstanding_wait_time(&db).await?;
log_job_view(
@@ -938,7 +943,7 @@ impl<'a> GetQuery<'a> {
async fn fetch_queued(
self,
db: &DB,
job_id: Uuid,
job_id: &Uuid,
workspace_id: &str,
) -> error::Result<Option<JobExtended<QueuedJob>>> {
let query = get_job_query!("v2_job_queue",
@@ -970,7 +975,7 @@ impl<'a> GetQuery<'a> {
async fn fetch_completed(
self,
db: &DB,
job_id: Uuid,
job_id: &Uuid,
workspace_id: &str,
) -> error::Result<Option<JobExtended<CompletedJob>>> {
let query = get_job_query!("v2_job_completed",
@@ -1007,7 +1012,7 @@ impl<'a> GetQuery<'a> {
Ok(cjob)
}
async fn fetch(self, db: &DB, job_id: Uuid, workspace_id: &str) -> error::Result<Job> {
async fn fetch(self, db: &DB, job_id: &Uuid, workspace_id: &str) -> error::Result<Job> {
let cjob = self
.fetch_completed(db.into(), job_id, workspace_id)
.await?
@@ -2092,7 +2097,7 @@ async fn resume_suspended_job_internal(
.without_logs()
.without_code()
.without_flow()
.fetch(&db, parent_flow_info.id, &w_id)
.fetch(&db, &parent_flow_info.id, &w_id)
.await?;
let flow_status = parent_flow
.flow_status()
@@ -2390,7 +2395,7 @@ pub async fn get_suspended_job_flow(
let flow = GetQuery::new()
.without_logs()
.without_code()
.fetch(&db, flow_id, &w_id)
.fetch(&db, &flow_id, &w_id)
.await?;
let flow_status = flow
@@ -3800,7 +3805,7 @@ pub async fn run_workflow_as_code(
let job = GetQuery::new()
.without_logs()
.fetch_queued(&db, job_id, &w_id)
.fetch_queued(&db, &job_id, &w_id)
.await?;
if *CLOUD_HOSTED {
@@ -5682,6 +5687,7 @@ pub struct JobUpdate {
pub mem_peak: Option<i32>,
pub progress: Option<i32>,
pub flow_status: Option<Box<serde_json::value::RawValue>>,
pub job: Option<Job>,
}
async fn get_log_file(Path((_w_id, file_p)): Path<(String, String)>) -> error::Result<Response> {
@@ -5751,6 +5757,7 @@ async fn get_job_update(
log_offset,
get_progress,
running,
false,
).await?))
}
@@ -5810,10 +5817,15 @@ fn get_job_update_sse_stream(
log_offset,
get_progress,
running,
true
).await {
if update.completed.unwrap_or(false) {
return;
}
if let Ok(serialized) = serde_json::to_string(&update) {
let event_data = format!("data: {}\n\n", serialized);
if tx.send(event_data.clone()).await.is_err() {
tracing::warn!("Failed to send initial job update for job {job_id}");
return;
}
last_update = Some(serialized);
@@ -5830,9 +5842,11 @@ fn get_job_update_sse_stream(
}
// Poll for updates every 1 second
let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
let mut i = 0;
loop {
interval.tick().await;
i += 1;
let ms_duration = if i > 10 { 1000 } else if i > 50 { 5000 } else { 100 };
tokio::time::sleep(std::time::Duration::from_millis(ms_duration)).await;
match get_job_update_data(
&opt_authed,
@@ -5843,6 +5857,7 @@ fn get_job_update_sse_stream(
log_offset,
get_progress,
running,
true,
).await {
Ok(update) => {
if let Ok(serialized) = serde_json::to_string(&update) {
@@ -5852,6 +5867,9 @@ fn get_job_update_sse_stream(
if tx.send(event_data).await.is_err() {
break;
}
if update.completed.unwrap_or(false) {
break;
}
last_update = Some(serialized);
// Update log offset if available
@@ -5859,21 +5877,6 @@ fn get_job_update_sse_stream(
log_offset = new_offset;
}
// Check if job is completed
if update.completed.unwrap_or(false) {
// let full_job = get_job_by_id(db, &job_id).await?;
let mut get = GetQuery::new()
.with_auth(&opt_authed)
.without_logs();
let full_job = get.fetch(&db, job_id, &w_id).await?;
tx.send(
format!("data: {{'completed': true, 'job': {}}}\n\n",
serde_json::to_string(&full_job).unwrap())).await.unwrap();
break;
}
}
}
}
@@ -5897,6 +5900,7 @@ async fn get_job_update_data(
log_offset: i32,
get_progress: Option<bool>,
running: bool,
get_full_job_on_completion: bool,
) -> error::Result<JobUpdate> {
let record = sqlx::query!(
"SELECT
@@ -5960,6 +5964,15 @@ async fn get_job_update_data(
)
.await?;
let job = if record.completed.unwrap_or(false) && get_full_job_on_completion {
let get = GetQuery::new()
.with_auth(&opt_authed)
.without_logs();
Some(get.fetch(&db, job_id, &w_id).await?)
} else {
None
};
Ok(JobUpdate {
running: record.running,
completed: record.completed,
@@ -5967,6 +5980,7 @@ async fn get_job_update_data(
new_logs: record.logs,
mem_peak: record.mem_peak,
progress: record.progress,
job,
flow_status: record
.flow_status
.map(|x: sqlx::types::Json<Box<RawValue>>| x.0),
@@ -6259,7 +6273,7 @@ async fn get_completed_job<'a>(
let job_o = GetQuery::new()
.with_auth(&opt_authed)
.with_in_tags(tags.as_ref())
.fetch_completed(&db, id, &w_id)
.fetch_completed(&db, &id, &w_id)
.await?;
let cj = not_found_if_none(job_o, "Completed Job", id.to_string())?;
frontend/src/lib/components/Dev.svelte
+1
View File
@@ -638,6 +638,7 @@
<svelte:window onkeydown={onKeyDown} />
<TestJobLoader
noCode={true}
on:done={loadPastTests}
bind:this={testJobLoader}
bind:isLoading={testIsLoading}
frontend/src/lib/components/LogViewer.svelte
+68 -47
View File
@@ -1,4 +1,4 @@
<script lang="ts" context="module">
<script lang="ts" module>
const s3LogPrefixes = [
'\n[windmill] Previous logs have been saved to object storage at logs/',
'\n[windmill] Previous logs have been saved to disk at logs/',
@@ -17,41 +17,54 @@
import { JobService } from '$lib/gen'
import Tooltip from './Tooltip.svelte'
export let content: string | undefined
export let isLoading: boolean
export let duration: number | undefined = undefined
export let mem: number | undefined = undefined
export let wrapperClass = ''
export let jobId: string | undefined = undefined
export let tag: string | undefined
export let small = false
export let drawerOpen = false
export let noMaxH = false
export let noAutoScroll = false
export let download = true
export let customEmptyMessage = 'No logs are available yet'
interface Props {
content: string | undefined
isLoading: boolean
duration?: number | undefined
mem?: number | undefined
wrapperClass?: string
jobId?: string | undefined
tag: string | undefined
small?: boolean
drawerOpen?: boolean
noMaxH?: boolean
noAutoScroll?: boolean
download?: boolean
customEmptyMessage?: string
}
let {
content,
isLoading,
duration = undefined,
mem = undefined,
wrapperClass = '',
jobId = undefined,
tag,
small = false,
drawerOpen = $bindable(false),
noMaxH = false,
noAutoScroll = false,
download = true,
customEmptyMessage = 'No logs are available yet'
}: Props = $props()
// @ts-ignore
const ansi_up = new AnsiUp()
const ansi_up = $state(new AnsiUp())
ansi_up.use_classes = true
let scroll = true
let div: HTMLElement | null = null
let scroll = $state(true)
let div: HTMLElement | null = $state(null)
// let downloadStartUrl: string | undefined = undefined
let LOG_INC = 10000
let LOG_LIMIT = LOG_INC
let LOG_LIMIT = $state(LOG_INC)
let lastJobId = jobId
let lastJobId = $state(jobId)
let loadedFromObjectStore = ''
$: if (jobId !== lastJobId) {
lastJobId = jobId
loadedFromObjectStore = ''
LOG_LIMIT = LOG_INC
}
let loadedFromObjectStore = $state('')
function findPrefixIndex(truncateContent: string): number | undefined {
let index = s3LogPrefixes.findIndex((x) => truncateContent.startsWith(x))
@@ -85,11 +98,6 @@
}
}
$: truncatedContent = truncateContent(content, loadedFromObjectStore, LOG_LIMIT)
$: prefixIndex = findPrefixIndex(truncatedContent)
$: downloadStartUrl = findStartUrl(truncatedContent, prefixIndex)
function truncateContent(
jobContent: string | undefined,
loadedFromObjectStore: string,
@@ -102,21 +110,12 @@
return content
}
$: truncatedContent && scrollToBottom()
$: html = ansi_up.ansi_to_html(
downloadStartUrl && prefixIndex != undefined
? truncatedContent.substring(
truncatedContent.substring(1).indexOf('\n') + 2,
truncatedContent.length
)
: truncatedContent
)
export function scrollToBottom() {
// console.log('scrollToBottom', scroll, div)
scroll && setTimeout(() => div?.scroll({ top: div?.scrollHeight, behavior: 'smooth' }), 100)
}
let logViewer: Drawer
let logViewer: Drawer | undefined = $state()
async function getStoreLogs() {
if (downloadStartUrl) {
@@ -138,13 +137,35 @@
function showMoreTruncate(len: number) {
scroll = false
LOG_LIMIT += LOG_INC
console.log(LOG_INC, len, LOG_LIMIT)
let newC = truncateContent(content, loadedFromObjectStore, LOG_LIMIT)
let newlineIndex = newC.indexOf('\n') + 1
if (newlineIndex < LOG_INC / 2) {
LOG_LIMIT -= newlineIndex
}
}
$effect.pre(() => {
if (jobId !== lastJobId) {
lastJobId = jobId
loadedFromObjectStore = ''
LOG_LIMIT = LOG_INC
}
})
let truncatedContent = $derived(truncateContent(content, loadedFromObjectStore, LOG_LIMIT))
let prefixIndex = $derived(findPrefixIndex(truncatedContent))
let downloadStartUrl = $derived(findStartUrl(truncatedContent, prefixIndex))
$effect.pre(() => {
truncatedContent && scrollToBottom()
})
let html = $derived(
ansi_up.ansi_to_html(
downloadStartUrl && prefixIndex != undefined
? truncatedContent.substring(
truncatedContent.substring(1).indexOf('\n') + 2,
truncatedContent.length
)
: truncatedContent
)
)
</script>
<Drawer bind:this={logViewer} bind:open={drawerOpen} size="900px">
@@ -181,10 +202,10 @@
>{#if content}{@const len =
(content?.length ?? 0) +
(loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}<button
on:click={getStoreLogs}
onclick={getStoreLogs}
>Show more... <Tooltip>{tooltipText(prefixIndex)}</Tooltip></button
><br />{:else if len > LOG_LIMIT}(truncated to the last {LOG_LIMIT} characters)...<br
/><button on:click={() => showMoreTruncate(len)}>Show more..</button><br
/><button onclick={() => showMoreTruncate(len)}>Show more..</button><br
/>{/if}{@html html}{:else if isLoading}Waiting for job to start...{:else}No logs are available yet{/if}</pre
>
</div>
@@ -209,7 +230,7 @@
</a>
</div>
{/if}
<button on:click={logViewer.openDrawer}><Expand size="12" /></button>
<button onclick={logViewer.openDrawer}><Expand size="12" /></button>
{#if !noAutoScroll}
<div
class="{small ? '' : 'py-2'} pr-2 {small
@@ -249,10 +270,10 @@
<pre class="whitespace-pre break-words {small ? '!text-2xs' : '!text-xs'} w-full p-2"
>{#if content}{@const len =
(content?.length ?? 0) +
(loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}<button on:click={getStoreLogs}
(loadedFromObjectStore?.length ?? 0)}{#if downloadStartUrl}<button onclick={getStoreLogs}
>Show more... &nbsp;<Tooltip>{tooltipText(prefixIndex)}</Tooltip></button
><br />{:else if len > LOG_LIMIT}(truncated to the last {LOG_LIMIT} characters)<br
/><button on:click={() => showMoreTruncate(len)}>Show more..</button><br />{/if}<span
/><button onclick={() => showMoreTruncate(len)}>Show more..</button><br />{/if}<span
>{@html html}</span
>{:else if !isLoading}<span>{customEmptyMessage}</span>{/if}</pre
>
frontend/src/lib/components/ModuleTest.svelte
+1
View File
@@ -117,6 +117,7 @@
</script>
<TestJobLoader
noCode={true}
toastError={noEditor}
on:done={() => jobDone()}
bind:scriptProgress
frontend/src/lib/components/ScriptEditor.svelte
+1
View File
@@ -445,6 +445,7 @@
</script>
<TestJobLoader
noCode={true}
on:done={loadPastTests}
bind:scriptProgress
bind:this={testJobLoader}
frontend/src/lib/components/TestJobLoader.svelte
+10 -6
View File
@@ -12,6 +12,7 @@
interface Props {
isLoading?: boolean
job?: Job | undefined
noCode?: boolean
workspaceOverride?: string | undefined
notfound?: boolean
jobUpdateLastFetch?: Date | undefined
@@ -25,6 +26,7 @@
let {
isLoading = $bindable(false),
job = $bindable(undefined),
noCode = false,
workspaceOverride = undefined,
notfound = $bindable(false),
jobUpdateLastFetch = $bindable(undefined),
@@ -286,10 +288,10 @@
job.mem_peak = previewJobUpdates.mem_peak
}
if ((previewJobUpdates.running ?? false) || (previewJobUpdates.completed ?? false)) {
job = await JobService.getJob({ workspace: workspace!, id })
job = await JobService.getJob({ workspace: workspace!, id, noCode })
}
} else {
job = await JobService.getJob({ workspace: workspace!, id, noLogs: lazyLogs })
job = await JobService.getJob({ workspace: workspace!, id, noLogs: lazyLogs, noCode })
}
jobUpdateLastFetch = new Date()
@@ -325,7 +327,7 @@
try {
// First load the job to get initial state
if (!job) {
job = await JobService.getJob({ workspace: workspace!, id, noLogs: lazyLogs })
job = await JobService.getJob({ workspace: workspace!, id, noLogs: lazyLogs, noCode })
}
// If job is already completed, don't start SSE
@@ -393,6 +395,8 @@
scriptProgress = clamp(previewJobUpdates.progress, scriptProgress ?? 0, 99)
}
console.log('previewJobUpdates', previewJobUpdates.log_offset)
if (previewJobUpdates.new_logs && job) {
if (offset == 0) {
job.logs = previewJobUpdates.new_logs ?? ''
@@ -414,7 +418,9 @@
// Check if job is completed
if (previewJobUpdates.completed) {
job = await JobService.getJob({ workspace: workspace!, id })
const njob = previewJobUpdates.job as Job
njob.logs = job?.logs ?? ''
job = njob
currentEventSource?.close()
currentEventSource = undefined
@@ -426,8 +432,6 @@
currentId = undefined
}
}
} else if (previewJobUpdates.running ?? false) {
job = await JobService.getJob({ workspace: workspace!, id })
}
} catch (parseErr) {
console.warn('Failed to parse SSE data:', parseErr)
frontend/src/lib/components/apps/components/display/AppDisplayComponentByJobId.svelte
+1
View File
@@ -90,6 +90,7 @@
{/each}
<TestJobLoader
noCode={true}
workspaceOverride={workspace}
bind:this={testJobLoader}
bind:isLoading={testIsLoading}
frontend/src/lib/components/apps/components/display/AppJobIdLogComponent.svelte
+1
View File
@@ -87,6 +87,7 @@
{/each}
<TestJobLoader
noCode={true}
workspaceOverride={workspace}
bind:this={testJobLoader}
bind:isLoading={testIsLoading}
frontend/src/lib/components/apps/editor/AppJobsDrawer.svelte
+1 -1
View File
@@ -63,7 +63,7 @@
})
</script>
<TestJobLoader bind:this={testJobLoader} bind:isLoading={testIsLoading} bind:job />
<TestJobLoader noCode={true} bind:this={testJobLoader} bind:isLoading={testIsLoading} bind:job />
<Drawer bind:open size="900px">
<DrawerContent
frontend/src/lib/components/raw_apps/RawAppInlineScriptRunnable.svelte
+1
View File
@@ -95,6 +95,7 @@
</script>
<TestJobLoader
noCode={true}
bind:scriptProgress
bind:this={testJobLoader}
bind:isLoading={testIsLoading}