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

feat: add flow debug info endpoint + button

Browse Source
This commit is contained in:
Ruben Fiszel
2024-02-04 22:16:12 +01:00
parent 6d33964227
commit 2568ea0ee2
7 changed files with 190 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/openapi.yaml
+17
View File
@@ -5032,6 +5032,23 @@ paths:
mem_peak:
type: integer
/w/{workspace}/jobs_u/get_flow_debug_info/{id}:
get:
summary: get flow debug info
operationId: getFlowDebugInfo
tags:
- job
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/JobId"
responses:
"200":
description: flow debug info details
content:
application/json:
schema: {}
/w/{workspace}/jobs_u/completed/get/{id}:
get:
summary: get completed job
backend/windmill-api/src/jobs.rs
+69 -4
View File
@@ -11,7 +11,7 @@ use serde_json::value::RawValue;
use std::collections::HashMap;
use std::sync::atomic::Ordering;
use tokio::time::Instant;
use windmill_common::flow_status::RestartedFrom;
use windmill_common::flow_status::{JobResult, RestartedFrom};
use windmill_common::variables::get_workspace_key;
use crate::db::ApiAuthed;
@@ -206,6 +206,7 @@ pub fn global_service() -> Router {
)
.route("/get/:id", get(get_job))
.route("/get_logs/:id", get(get_job_logs))
.route("/get_flow_debug_info/:id", get(get_flow_job_debug_info))
.route("/completed/get/:id", get(get_completed_job))
.route("/completed/get_result/:id", get(get_completed_job_result))
.route(
@@ -436,6 +437,59 @@ pub async fn get_path_tag_limits_cache_for_hash(
))
}
async fn get_flow_job_debug_info(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
) -> error::Result<Response> {
let job = get_queued_job(id, w_id.as_str(), &db).await?;
if let Some(job) = job {
let is_flow = &job.job_kind == &JobKind::FlowPreview || &job.job_kind == &JobKind::Flow;
if job.is_flow_step || !is_flow {
return Err(error::Error::BadRequest(
"This endpoint is only for root flow jobs".to_string(),
));
}
let mut jobs = HashMap::new();
jobs.insert("root_job".to_string(), job.clone());
let mut job_ids = vec![];
let jobs_with_root = sqlx::query_scalar!(
"SELECT id FROM queue WHERE workspace_id = $1 and root_job = $2",
&w_id,
&job.id,
)
.fetch_all(&db)
.await?;
for job in jobs_with_root {
job_ids.push(job);
}
let leaf_jobs: HashMap<String, JobResult> = job
.leaf_jobs
.and_then(|x| serde_json::from_value(x).ok())
.unwrap_or_else(HashMap::new);
for job in leaf_jobs.iter() {
match job.1 {
JobResult::ListJob(jobs) => job_ids.extend(jobs.to_owned()),
JobResult::SingleJob(job) => job_ids.push(job.clone()),
}
}
for job_id in job_ids {
let job = get_queued_job(job_id, w_id.as_str(), &db).await?;
if let Some(job) = job {
jobs.insert(job.id.to_string(), job);
}
}
Ok(Json(jobs).into_response())
} else {
Err(error::Error::NotFound(format!(
"QueuedJob {} not found",
id
)))
}
}
async fn get_job(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
@@ -749,9 +803,7 @@ async fn cancel_all(
for j in jobs.iter() {
if !j.running && !j.is_flow_step.unwrap_or(false) {
let e = serde_json::json!({"message": format!("Job canceled: cancel_all by {username}"), "name": "Canceled", "reason": "cancel_all", "canceler": username});
let mut tx = db.begin().await?;
let job_running = get_queued_job(j.id, &w_id, &mut tx).await?;
tx.commit().await?;
let job_running = get_queued_job(j.id, &w_id, &db).await?;
if let Some(job_running) = job_running {
let add_job = add_completed_job_error(
@@ -1497,6 +1549,19 @@ impl Job {
.flatten(),
}
}
pub fn is_flow_step(&self) -> bool {
match self {
Job::QueuedJob(job) => job.is_flow_step,
Job::CompletedJob(job) => job.is_flow_step,
}
}
pub fn job_kind(&self) -> &JobKind {
match self {
Job::QueuedJob(job) => &job.job_kind,
Job::CompletedJob(job) => &job.job_kind,
}
}
}
#[derive(sqlx::FromRow)]
backend/windmill-api/src/oidc.rs
+1 -3
View File
@@ -208,9 +208,7 @@ pub async fn gen_token(
job.unwrap()
}
};
let mut tx = db.begin().await?;
let job = get_queued_job(job_id, &w_id, &mut tx).await?;
tx.commit().await?;
let job = get_queued_job(job_id, &w_id, &db).await?;
let job = job.ok_or_else(|| anyhow::anyhow!("Queued job {} not found", job_id))?;
let issue_url = format!("{}/api/oidc/", crate::BASE_URL.read().await.clone());
backend/windmill-queue/src/jobs.rs
+18 -2
View File
@@ -131,7 +131,7 @@ pub async fn cancel_job<'c: 'async_recursion>(
rsmq: Option<rsmq_async::MultiplexedRsmq>,
force_cancel: bool,
) -> error::Result<(Transaction<'c, Postgres>, Option<Uuid>)> {
let job_running = get_queued_job(id, &w_id, &mut tx).await?;
let job_running = get_queued_job_tx(id, &w_id, &mut tx).await?;
if job_running.is_none() {
return Ok((tx, None));
@@ -2166,7 +2166,7 @@ pub async fn job_is_complete(db: &DB, id: Uuid, w_id: &str) -> error::Result<boo
.unwrap_or(false))
}
pub async fn get_queued_job<'c>(
pub async fn get_queued_job_tx<'c>(
id: Uuid,
w_id: &str,
tx: &mut Transaction<'c, Postgres>,
@@ -2186,6 +2186,22 @@ pub async fn get_queued_job<'c>(
}
}
pub async fn get_queued_job(id: Uuid, w_id: &str, db: &DB) -> error::Result<Option<QueuedJob>> {
let r = sqlx::query(
"SELECT *
FROM queue WHERE id = $1 AND workspace_id = $2",
)
.bind(id)
.bind(w_id)
.fetch_optional(db)
.await?;
if let Some(row) = r {
Ok(Some(QueuedJob::from_row(&row)?.to_owned()))
} else {
Ok(None)
}
}
pub enum PushIsolationLevel<'c, R: rsmq_async::RsmqConnection + Send + 'c> {
IsolatedRoot(DB, Option<R>),
Isolated(UserDB, Authed, Option<R>),
backend/windmill-worker/src/worker.rs
+15 -17
View File
@@ -2265,23 +2265,21 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
if let Err(err) = updated_flow {
if let Some(parent_job_id) = job.parent_job {
if let Ok(mut tx) = db.begin().await {
if let Ok(Some(parent_job)) =
get_queued_job(parent_job_id, &job.workspace_id, &mut tx).await
{
let e = json!({"message": err.to_string(), "name": "InternalErr"});
let _ = add_completed_job_error(
db,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
canceled_by.clone(),
e,
rsmq,
worker_name,
)
.await;
}
if let Ok(Some(parent_job)) =
get_queued_job(parent_job_id, &job.workspace_id, &db).await
{
let e = json!({"message": err.to_string(), "name": "InternalErr"});
let _ = add_completed_job_error(
db,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
canceled_by.clone(),
e,
rsmq,
worker_name,
)
.await;
}
}
}
backend/windmill-worker/src/worker_flow.rs
+5 -7
View File
@@ -43,13 +43,13 @@ use windmill_common::{
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::{
add_completed_job, add_completed_job_error, handle_maybe_scheduled_job, CanceledBy,
PushIsolationLevel, WrappedError,
add_completed_job, add_completed_job_error, get_queued_job, handle_maybe_scheduled_job,
CanceledBy, PushIsolationLevel, WrappedError,
};
type DB = sqlx::Pool<sqlx::Postgres>;
use windmill_queue::{canceled_job_to_result, get_queued_job, push, QueueTransaction};
use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction};
// #[instrument(level = "trace", skip_all)]
pub async fn update_flow_status_after_job_completion<
@@ -231,11 +231,9 @@ pub async fn update_flow_status_after_job_completion_internal<
let (mut stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override {
//do not stop early if module is a flow step
let mut tx = db.begin().await?;
let flow_job = get_queued_job(flow, w_id, &mut tx)
let flow_job = get_queued_job(flow, w_id, db)
.await?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
tx.commit().await?;
let module = get_module(&flow_job, module_index);
if module.is_some_and(|x| matches!(x.value, FlowModuleValue::Flow { .. })) {
(false, false)
@@ -524,7 +522,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.context("remove flow status retry")?;
}
let flow_job = get_queued_job(flow, w_id, tx.transaction_mut())
let flow_job = get_queued_job_tx(flow, w_id, tx.transaction_mut())
.await?
.ok_or_else(|| Error::InternalErr(format!("requiring flow to be in the queue")))?;
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+65 -3
View File
@@ -1,7 +1,7 @@
<script lang="ts">
import { page } from '$app/stores'
import { JobService, Job, ScriptService, Script } from '$lib/gen'
import { canWrite, displayDate, emptyString, truncateHash } from '$lib/utils'
import { canWrite, copyToClipboard, displayDate, emptyString, truncateHash } from '$lib/utils'
import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte'
import {
@@ -19,7 +19,9 @@
TimerOff,
Trash,
XCircle,
Code2
Code2,
ClipboardCopy,
MoreVertical
} from 'lucide-svelte'
import DisplayResult from '$lib/components/DisplayResult.svelte'
@@ -36,7 +38,16 @@
import HighlightCode from '$lib/components/HighlightCode.svelte'
import TestJobLoader from '$lib/components/TestJobLoader.svelte'
import LogViewer from '$lib/components/LogViewer.svelte'
import { ActionRow, Button, Popup, Skeleton, Tab, Alert, MenuItem } from '$lib/components/common'
import {
ActionRow,
Button,
Popup,
Skeleton,
Tab,
Alert,
MenuItem,
DrawerContent
} from '$lib/components/common'
import FlowMetadata from '$lib/components/FlowMetadata.svelte'
import JobArgs from '$lib/components/JobArgs.svelte'
import FlowProgressBar from '$lib/components/flows/FlowProgressBar.svelte'
@@ -50,6 +61,9 @@
import PersistentScriptDrawer from '$lib/components/PersistentScriptDrawer.svelte'
import Portal from 'svelte-portal'
import MemoryFootprintViewer from '$lib/components/MemoryFootprintViewer.svelte'
import Drawer from '$lib/components/common/drawer/Drawer.svelte'
import { Highlight } from 'svelte-highlight'
import { json } from 'svelte-highlight/languages'
let job: Job | undefined
let jobUpdateLastFetch: Date | undefined
@@ -166,8 +180,40 @@
let notfound = false
let forceCancel = false
let debugViewer: Drawer
let debugContent: any = undefined
async function debugInfo() {
if (job?.id) {
debugContent = await JobService.getFlowDebugInfo({ workspace: $workspaceStore!, id: job?.id })
debugViewer?.openDrawer()
} else {
sendUserToast('Job has no id', true)
}
}
</script>
{#if (job?.job_kind == 'flow' || job?.job_kind == 'flowpreview') && job?.['running'] && job?.parent_job == undefined}
<Drawer bind:this={debugViewer} size="800px">
<DrawerContent title="Debug Detail" on:close={debugViewer.closeDrawer}>
<svelte:fragment slot="actions">
<Button
on:click={() => copyToClipboard(JSON.stringify(debugContent, null, 4))}
color="light"
size="xs"
>
<div class="flex gap-2 items-center">Copy <ClipboardCopy /> </div>
</Button>
</svelte:fragment>
<pre
><code class="text-2xs p-2">
<Highlight language={json} code={JSON.stringify(debugContent, null, 4)} />
</code></pre
>
</DrawerContent>
</Drawer>
{/if}
<TestJobLoader
on:done={() => (viewTab = 'result')}
bind:this={testJobLoader}
@@ -256,6 +302,22 @@
{@const stem = `/${job?.job_kind}s`}
{@const isScript = job?.job_kind === 'script'}
{@const viewHref = `${stem}/get/${isScript ? job?.script_hash : job?.script_path}`}
{#if (job?.job_kind == 'flow' || job?.job_kind == 'flowpreview') && job?.['running'] && job?.parent_job == undefined}
<div class="inline">
<ButtonDropdown hasPadding={false}>
<svelte:fragment slot="buttonReplacement">
<Button nonCaptureEvent size="xs" color="light">
<div class="flex flex-row items-center">
<MoreVertical size={14} />
</div>
</Button>
</svelte:fragment>
<svelte:fragment slot="items">
<MenuItem on:click={debugInfo}>Show Flow Debug Info</MenuItem>
</svelte:fragment>
</ButtonDropdown>
</div>
{/if}
{#if persistentScriptDefinition !== undefined}
<Button
color="blue"