kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

early stop now bubble up to the top

Browse Source
This commit is contained in:
Ruben Fiszel
2022-10-30 02:32:35 +02:00
parent bb4a620645
commit e96c5ca670
4 changed files with 37 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/windmill-worker/src/worker.rs
View File

@@ -503,6 +503,7 @@ async fn handle_job_error(
worker_dir,
keep_job_dir,
base_internal_url,
None,
)
.await;
if let Err(err) = updated_flow {
@@ -656,6 +657,7 @@ async fn handle_queued_job(
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
None,
)
.await?;
}
@@ -705,6 +707,7 @@ async fn handle_queued_job(
worker_dir,
worker_config.keep_job_dir,
&worker_config.base_internal_url,
None,
)
.await?;
}

37
backend/windmill-worker/src/worker_flow.rs
View File

@@ -50,6 +50,7 @@ pub async fn update_flow_status_after_job_completion(
worker_dir: &str,
keep_job_dir: bool,
base_internal_url: &str,
stop_early_override: Option<bool>,
) -> error::Result<()> {
tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id}");
@@ -152,8 +153,13 @@ pub async fn update_flow_status_after_job_completion(
.map(|i| !(..old_status.modules.len()).contains(&i))
.unwrap_or(true);
let (stop_early_expr, skip_if_stop_early) =
sqlx::query_as::<_, (Option<String>, Option<bool>)>(
let (stop_early, skip_if_stop_early) = if let Some(se) = stop_early_override {
(true, se)
} else {
let (stop_early_expr, skip_if_stop_early) = sqlx::query_as::<
_,
(Option<String>, Option<bool>),
>(
"
UPDATE queue
SET flow_status = JSONB_SET(
@@ -173,12 +179,16 @@ pub async fn update_flow_status_after_job_completion(
.await
.map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?;
let stop_early = success
&& if let Some(expr) = stop_early_expr.clone() {
compute_bool_from_expr(expr, result.clone(), base_internal_url).await?
} else {
false
};
(
success
&& if let Some(expr) = stop_early_expr.clone() {
compute_bool_from_expr(expr, result.clone(), base_internal_url).await?
} else {
false
},
skip_if_stop_early.unwrap_or(false),
)
};
let result = match &new_status {
FlowStatusModule::Success { flow_jobs: Some(jobs), .. } => {
@@ -266,8 +276,7 @@ pub async fn update_flow_status_after_job_completion(
let logs = if flow_job.canceled {
"Flow job canceled".to_string()
} else if stop_early {
let stop_early_expr = stop_early_expr.unwrap();
format!("Flow job stopped early based on the stop_early_expr predicate: {stop_early_expr} returning true")
format!("Flow job stopped early because of a stop early predicate returning true")
} else {
"Flow job completed".to_string()
};
@@ -288,7 +297,7 @@ pub async fn update_flow_status_after_job_completion(
client,
&flow_job,
success,
stop_early && skip_if_stop_early.unwrap_or(false),
stop_early && skip_if_stop_early,
result.clone(),
logs,
)
@@ -343,6 +352,11 @@ pub async fn update_flow_status_after_job_completion(
worker_dir,
keep_job_dir,
base_internal_url,
if stop_early {
Some(skip_if_stop_early)
} else {
None
},
)
.await?);
}
@@ -589,6 +603,7 @@ pub async fn handle_flow(
worker_dir,
false,
base_internal_url,
None,
)
.await?;
return Ok(());

8
frontend/src/lib/components/flows/content/FlowBranchesWrapper.svelte
View File

@@ -6,7 +6,7 @@
import FlowCard from '../common/FlowCard.svelte'
import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte'
import FlowModuleSuspend from './FlowModuleSuspend.svelte'
import FlowRetries from './FlowRetries.svelte'
// import FlowRetries from './FlowRetries.svelte'
export let flowModule: FlowModule
export let type: 'branchall' | 'branchone'
@@ -50,17 +50,17 @@
</div>
{#if flowModule}
<Tabs bind:selected>
<Tab value="retries">Retries</Tab>
<!-- <Tab value="retries">Retries</Tab> -->
<Tab value="early-stop">Early Stop</Tab>
<Tab value="suspend">Sleep/Suspend</Tab>
<svelte:fragment slot="content">
<div class="overflow-hidden bg-white">
<TabContent value="retries" class="flex flex-col flex-1 h-full">
<!-- <TabContent value="retries" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowRetries bind:flowModule />
</div>
</TabContent>
</TabContent> -->
<TabContent value="early-stop" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">

8
frontend/src/lib/components/flows/content/FlowLoop.svelte
View File

@@ -8,7 +8,7 @@
import PropPickerWrapper from '../propPicker/PropPickerWrapper.svelte'
import FlowModuleEarlyStop from './FlowModuleEarlyStop.svelte'
import FlowModuleSuspend from './FlowModuleSuspend.svelte'
import FlowRetries from './FlowRetries.svelte'
// import FlowRetries from './FlowRetries.svelte'
import { Button, Tab, TabContent, Tabs } from '$lib/components/common'
import type { FlowModule } from '$lib/gen/models/FlowModule'
import { Pane, Splitpanes } from 'svelte-splitpanes'
@@ -85,17 +85,17 @@
</Pane>
<Pane size={40} minSize={20} class="flex flex-col flex-1">
<Tabs bind:selected>
<Tab value="retries">Retries</Tab>
<!-- <Tab value="retries">Retries</Tab> -->
<Tab value="early-stop">Early Stop</Tab>
<Tab value="suspend">Sleep/Suspend</Tab>
<svelte:fragment slot="content">
<div class="overflow-hidden bg-white" style="height:calc(100% - 32px);">
<TabContent value="retries" class="flex flex-col flex-1 h-full">
<!-- <TabContent value="retries" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">
<FlowRetries bind:flowModule={mod} />
</div>
</TabContent>
</TabContent> -->
<TabContent value="early-stop" class="flex flex-col flex-1 h-full">
<div class="p-4 overflow-y-auto">