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

error handling improvements

Browse Source
This commit is contained in:
Ruben Fiszel
2023-01-14 17:46:03 +01:00
parent 6949516e63
commit 776d4c9ca1
9 changed files with 81 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-queue/src/jobs.rs
+2 -2
View File
@@ -545,13 +545,13 @@ pub async fn push<'c>(
Ok((uuid, tx))
}
pub fn canceled_job_to_result(job: &QueuedJob) -> String {
pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value {
let reason = job
.canceled_reason
.as_deref()
.unwrap_or_else(|| "no reason given");
let canceler = job.canceled_by.as_deref().unwrap_or_else(|| "unknown");
format!("Job canceled: {reason} by {canceler}")
serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler})
}
pub async fn get_hub_script(path: String, email: &str) -> error::Result<HubScript> {
backend/windmill-worker/src/jobs.rs
+6 -15
View File
@@ -14,26 +14,17 @@ use windmill_common::{error::Error, flow_status::FlowStatusModule, schedule::Sch
use windmill_queue::{delete_job, schedule::get_schedule_opt, JobKind, QueuedJob};
#[instrument(level = "trace", skip_all)]
pub async fn add_completed_job_error<E: ToString + std::fmt::Debug>(
pub async fn add_completed_job_error(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
logs: String,
e: E,
e: serde_json::Value,
metrics: Option<crate::worker::Metrics>,
) -> Result<(Uuid, serde_json::Map<String, serde_json::Value>), Error> {
) -> Result<serde_json::Value, Error> {
metrics.map(|m| m.worker_execution_failed.inc());
let mut output_map = Map::new();
error_to_result(&mut output_map, &e);
let a = add_completed_job(
db,
&queued_job,
false,
false,
serde_json::Value::Object(output_map.clone()),
logs,
)
.await?;
Ok((a, output_map))
let result = serde_json::json!({ "error": e });
let _ = add_completed_job(db, &queued_job, false, false, result.clone(), logs).await?;
Ok(result)
}
pub fn error_to_result<E: ToString + std::fmt::Debug>(
backend/windmill-worker/src/worker.rs
+42 -27
View File
@@ -25,7 +25,7 @@ use windmill_common::{
};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob};
use serde_json::{json, Map, Value};
use serde_json::{json, Value};
use tokio::{
fs::{metadata, symlink, DirBuilder, File},
@@ -651,16 +651,14 @@ async fn handle_job_error(
keep_job_dir: bool,
base_internal_url: &str,
) {
add_completed_job_error(
let _ = add_completed_job_error(
db,
&job,
format!("Unexpected error during job execution:\n{err}"),
&err,
json!({"message": err.to_string(), "name": "InternalErr"}),
metrics.clone(),
)
.await
.map(|(_, m)| m)
.unwrap_or_else(|_| Map::new());
.await;
if job.is_flow_step || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::Flow {
let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job {
@@ -700,7 +698,7 @@ async fn handle_job_error(
db,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
err,
json!({"message": err.to_string(), "name": "InternalErr"}),
metrics.clone(),
)
.await;
@@ -742,6 +740,14 @@ struct Envs {
max_log_size: i64,
}
fn extract_error_value(log_lines: &str) -> serde_json::Value {
if let Some(last) = log_lines.split("\n").last() {
if let Ok(v) = serde_json::from_str(last) {
return v;
};
}
return json!({"message": log_lines.to_string().trim().to_string(), "name": "ExecutionErr"});
}
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_queued_job(
job: QueuedJob,
@@ -759,7 +765,9 @@ async fn handle_queued_job(
base_internal_url: &str,
) -> windmill_common::error::Result<()> {
if job.canceled {
return Err(Error::ExecutionErr(canceled_job_to_result(&job)))?;
return Err(Error::ExecutionErr(
canceled_job_to_result(&job).to_string(),
))?;
}
if let Some(e) = job.pre_run_error {
return Err(Error::ExecutionErr(e));
@@ -862,7 +870,7 @@ async fn handle_queued_job(
}
}
Err(e) => {
let error_message = match e {
let error_value = match e {
Error::ExitStatus(_) => {
let last_10_log_lines = logs
.lines()
@@ -875,21 +883,17 @@ async fn handle_queued_job(
.split("CODE EXECUTION ---")
.last()
.unwrap_or(&logs);
log_lines.to_string().trim().to_string()
extract_error_value(log_lines)
}
err @ _ => {
json!({"message": format!("error before termination: {err:#?}"), "name": "ExecutionErr"})
}
err @ _ => format!("error before termination: {err:#?}"),
};
tracing::info!("job {} failed: {}", job.id, error_message);
let (_, output_map) = add_completed_job_error(
db,
&job,
logs,
error_message,
Some(metrics.clone()),
)
.await?;
let result =
add_completed_job_error(db, &job, logs, error_value, Some(metrics.clone()))
.await?;
if job.is_flow_step {
if let Some(parent_job) = job.parent_job {
update_flow_status_after_job_completion(
@@ -899,7 +903,7 @@ async fn handle_queued_job(
&job.id,
&job.workspace_id,
false,
serde_json::Value::Object(output_map),
result,
Some(metrics),
false,
same_worker_tx,
@@ -1454,7 +1458,10 @@ async function run() {{
await Deno.writeTextFile("result.json", res_json);
Deno.exit(0);
}}
run();
run().catch((e) => {{
console.error(JSON.stringify({{ message: e.message, name: e.name, stack: e.stack }}));
Deno.exit(1);
}});
"#,
);
write_file(job_dir, "main.ts", &wrapper_content).await?;
@@ -1686,6 +1693,8 @@ import json
{import_loader}
{import_base64}
{import_datetime}
import traceback
import sys
inner_script = __import__("inner")
@@ -1695,10 +1704,16 @@ for k, v in list(kwargs.items()):
if v == '<function call>':
del kwargs[k]
{transforms}
res = inner_script.main(**kwargs)
res_json = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
with open("result.json", 'w') as f:
f.write(res_json)
try:
res = inner_script.main(**kwargs)
res_json = json.dumps(res, separators=(',', ':'), default=str).replace('\n', '')
with open("result.json", 'w') as f:
f.write(res_json)
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
tb = traceback.format_tb(exc_traceback)
print(json.dumps({{ "message": str(e), "name": e.__class__.__name__, "stack": '\n'.join(tb[1:]) }}, separators=(',', ':'), default=str).replace('\n', ''), file=sys.stderr)
sys.exit(1)
"#,
);
write_file(job_dir, "main.py", &wrapper_content).await?;
backend/windmill-worker/src/worker_flow.rs
+4 -3
View File
@@ -402,7 +402,7 @@ pub async fn update_flow_status_after_job_completion(
{
true
}
false if has_failure_module(flow, &mut tx).await? => true,
false if has_failure_module(flow, &mut tx).await? && !is_failure_step => true,
false => false,
};
@@ -435,7 +435,7 @@ pub async fn update_flow_status_after_job_completion(
db,
&flow_job,
logs,
&canceled_job_to_result(&flow_job),
canceled_job_to_result(&flow_job),
metrics.clone(),
)
.await?;
@@ -468,7 +468,7 @@ pub async fn update_flow_status_after_job_completion(
db,
&flow_job,
"Unexpected error during flow chaining:\n".to_string(),
err,
json!({"message": err.to_string(), "name": "InternalError"}),
metrics.clone(),
)
.await;
@@ -692,6 +692,7 @@ async fn transform_input(
} else {
None
};
for (key, val) in input_transforms.into_iter() {
match val {
InputTransform::Static { value: _ } => (),
frontend/src/lib/components/ArgInput.svelte
-1
View File
@@ -157,7 +157,6 @@
}
}
export let inputCat: InputCat = 'string'
$: inputCat = computeInputCat(type, format, itemsType?.type, enum_, contentEncoding)
</script>
frontend/src/lib/components/DisplayResult.svelte
+4 -3
View File
@@ -59,7 +59,7 @@
return 'jpeg'
} else if (keys.length == 1 && keys[0] == 'file') {
return 'file'
} else if (keys.length == 1 && keys[0] == 'error' && typeof result['error'] == 'string') {
} else if (keys.length == 1 && keys[0] == 'error') {
return 'error'
} else if (
keys.length == 3 &&
@@ -141,8 +141,9 @@
>Download</a
>
</div>
{:else if !forceJson && resultKind == 'error'}<div
><pre class="text-sm text-red-500 whitespace-pre-wrap">{result.error}</pre>
{:else if !forceJson && resultKind == 'error'}<div>
<span class="text-red-500 font-semibold">{result.error.name}: {result.error.message}</span>
<pre class="text-sm whitespace-pre-wrap text-gray-900">{result.error.stack}</pre>
</div>
{:else if !forceJson && resultKind == 'approval'}<div class="flex flex-col gap-1 mx-4">
<Button
frontend/src/lib/components/InputTransformForm.svelte
+1 -2
View File
@@ -33,7 +33,7 @@
let monacoTemplate: TemplateEditor | undefined = undefined
let argInput: ArgInput | undefined = undefined
let inputCat: InputCat = computeInputCat(
$: inputCat = computeInputCat(
schema.properties[argName].type,
schema.properties[argName].format,
schema.properties[argName].items?.type,
@@ -269,7 +269,6 @@
bind:itemsType={schema.properties[argName].items}
properties={schema.properties[argName].properties}
displayHeader={false}
bind:inputCat
{variableEditor}
{itemPicker}
bind:pickForField
frontend/src/lib/components/propertyPicker/PropPicker.svelte
+7 -1
View File
@@ -101,7 +101,13 @@
<div class="overflow-y-auto mb-2">
<ObjectViewer
pureViewer={!$propPickerConfig}
json={{ error: 'The error to handle' }}
json={{
error: {
message: 'The error message',
name: 'The error name',
stack: 'The error stack'
}
}}
on:select
/>
</div>
frontend/src/lib/script_helpers.ts
+15 -15
View File
@@ -87,10 +87,11 @@ import (
// connect the error parameter to 'previous_result.error'
func main(error string) (interface{}, error) {
fmt.Println(error)
fmt.Println("job", os.Getenv("WM_JOB_ID"))
return x, nil
func main(message string, name string) (interface{}, error) {
fmt.Println(message)
fmt.Println(name)
fmt.Println("flow id that failed", os.Getenv("WM_FLOW_JOB_ID"))
return message, nil
}
`
@@ -102,13 +103,12 @@ export async function main(x: string) {
`
export const DENO_FAILURE_MODULE_CODE = `
// connect the error parameter to 'previous_result.error'
export async function main(error: string) {
const job = Deno.env.get("WM_JOB_ID")
console.log("error", error)
console.log("job", job)
return { error, job }
export async function main(message: string, name: string) {
const flow_id = Deno.env.get("WM_FLOW_JOB_ID")
console.log("message", message)
console.log("name",name)
return { message, flow_id }
}
`
@@ -122,11 +122,11 @@ export const PYTHON_FAILURE_MODULE_CODE = `import os
# connect the error parameter to 'previous_result.error'
def main(error: str):
job = os.environ.get("WM_JOB_ID")
print("error", error)
print("job", job)
return error, job
def main(message: str, name: str):
flow_id = os.environ.get("WM_FLOW_JOB_ID")
print("message", message)
print("name", name)
return message, flow_id
`
export const POSTGRES_INIT_CODE = `import {