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

feat: parse windmill_failure field to tag run as failure (#9073)

Browse Source 
* feat: parse windmill_failure field in job result to tag run as failure

* feat: preserve top-level fields when windmill_failure tags a run as failure

* fix: address review findings on windmill_manual_failure

* refactor: rename windmill_manual_failure to wm_failure and add wm_* aliases

* fix: prefer injected ManualFailure error over sibling name/message in OTel
This commit is contained in:
hugocasa
2026-05-08 17:57:34 +02:00
committed by GitHub
parent 23af6c2ea3
commit dd5320205f
3 changed files with 249 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-jobs/src/execution.rs
+3
View File
@@ -248,8 +248,11 @@ lazy_static::lazy_static! {
#[derive(Deserialize)]
pub struct WindmillCompositeResult {
#[serde(alias = "wm_status_code")]
windmill_status_code: Option<u16>,
#[serde(alias = "wm_content_type")]
windmill_content_type: Option<String>,
#[serde(alias = "wm_headers")]
windmill_headers: Option<HashMap<String, String>>,
result: Option<Box<RawValue>>,
}
backend/windmill-queue/src/jobs.rs
+137 -19
View File
@@ -630,12 +630,45 @@ pub struct WrappedError {
pub trait ValidableJson {
fn is_valid_json(&self) -> bool;
fn wm_labels(&self) -> Option<Vec<String>>;
fn wm_failure(&self) -> Option<String>;
fn result_metadata(&self) -> ResultMetadata;
fn size(&self) -> usize;
}
#[derive(serde::Deserialize)]
struct ResultLabels {
wm_labels: Vec<String>,
/// The Windmill-specific markers we look for inside a job's result.
/// `wm_failure` retags a successful run as a failure with the
/// given message; `wm_labels` adds runtime labels to the job row.
#[derive(serde::Deserialize, Default, Debug, Clone)]
pub struct ResultMetadata {
pub wm_labels: Option<Vec<String>>,
pub wm_failure: Option<String>,
}
/// Sentinel `error.name` we inject into a result when retagging a successful
/// run as a failure due to `wm_failure`. Used downstream to detect that
/// the result is already in the standard `{ error: { name, message }, ... }`
/// shape and must not be wrapped a second time by `WrappedError`.
pub const MANUAL_FAILURE_ERROR_NAME: &str = "ManualFailure";
/// Returns true when the result already carries our injected
/// `error: { name: "ManualFailure", ... }` marker — i.e. it was shaped by
/// `process_jc`'s wm_failure path. A real runtime failure whose raw
/// result happens to contain a `wm_failure` field but no such error
/// key returns false (and so still goes through the standard wrap path).
pub fn is_pre_shaped_wm_failure_result(result: &str) -> bool {
#[derive(serde::Deserialize)]
struct Marker {
error: Option<NameOnly>,
}
#[derive(serde::Deserialize)]
struct NameOnly {
name: String,
}
serde_json::from_str::<Marker>(result)
.ok()
.and_then(|m| m.error)
.map(|e| e.name == MANUAL_FAILURE_ERROR_NAME)
.unwrap_or(false)
}
impl ValidableJson for WrappedError {
@@ -647,6 +680,14 @@ impl ValidableJson for WrappedError {
None
}
fn wm_failure(&self) -> Option<String> {
None
}
fn result_metadata(&self) -> ResultMetadata {
ResultMetadata::default()
}
fn size(&self) -> usize {
0
}
@@ -658,9 +699,15 @@ impl ValidableJson for Box<RawValue> {
}
fn wm_labels(&self) -> Option<Vec<String>> {
serde_json::from_str::<ResultLabels>(self.get())
.ok()
.map(|r| r.wm_labels)
self.result_metadata().wm_labels
}
fn wm_failure(&self) -> Option<String> {
self.result_metadata().wm_failure
}
fn result_metadata(&self) -> ResultMetadata {
serde_json::from_str::<ResultMetadata>(self.get()).unwrap_or_default()
}
fn size(&self) -> usize {
@@ -677,6 +724,14 @@ impl<T: ValidableJson> ValidableJson for Arc<T> {
T::wm_labels(&self)
}
fn wm_failure(&self) -> Option<String> {
T::wm_failure(&self)
}
fn result_metadata(&self) -> ResultMetadata {
T::result_metadata(&self)
}
fn size(&self) -> usize {
T::size(&self)
}
@@ -688,9 +743,15 @@ impl ValidableJson for serde_json::Value {
}
fn wm_labels(&self) -> Option<Vec<String>> {
serde_json::from_value::<ResultLabels>(self.clone())
.ok()
.map(|r| r.wm_labels)
self.result_metadata().wm_labels
}
fn wm_failure(&self) -> Option<String> {
self.result_metadata().wm_failure
}
fn result_metadata(&self) -> ResultMetadata {
serde_json::from_value::<ResultMetadata>(self.clone()).unwrap_or_default()
}
fn size(&self) -> usize {
@@ -707,6 +768,14 @@ impl<T: ValidableJson> ValidableJson for Json<T> {
self.0.wm_labels()
}
fn wm_failure(&self) -> Option<String> {
self.0.wm_failure()
}
fn result_metadata(&self) -> ResultMetadata {
self.0.result_metadata()
}
fn size(&self) -> usize {
self.0.size()
}
@@ -742,16 +811,7 @@ where
}
}
pub async fn add_completed_job_error(
db: &Pool<Postgres>,
completed_job: &MiniCompletedJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
_worker_name: &str,
flow_is_done: bool,
duration: Option<i64>,
) -> Result<WrappedError, Error> {
async fn record_failure_metrics(completed_job: &MiniCompletedJob, _worker_name: &str) {
#[cfg(feature = "prometheus")]
register_metric(
&WORKER_EXECUTION_FAILED,
@@ -772,6 +832,64 @@ pub async fn add_completed_job_error(
.await;
otel_incr_worker_execution_failed(&completed_job.tag);
}
/// Tag a completed job as a failure while storing the result as-is, without
/// the standard `WrappedError` `{ error: ... }` wrap. Use for jobs whose result
/// is already shaped (e.g. when `wm_failure` injected a top-level
/// `error` key, while preserving sibling fields like `windmill_status_code`).
///
/// This is a worker-internal helper called by trusted result-processing code
/// after the worker has authenticated and pulled the job. Callers MUST verify
/// upstream auth (i.e. the job was legitimately pulled by this worker) — this
/// function performs no authorization check itself, mirroring the contract of
/// `add_completed_job_error`.
pub async fn add_completed_job_pre_shaped_failure<T: Serialize + Send + Sync + ValidableJson>(
db: &Pool<Postgres>,
completed_job: &MiniCompletedJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
result: Json<&T>,
worker_name: &str,
flow_is_done: bool,
duration: Option<i64>,
) -> Result<(), Error> {
record_failure_metrics(completed_job, worker_name).await;
tracing::error!(
"job {} in {} did not succeed (wm_failure)",
completed_job.id,
completed_job.workspace_id,
);
let _ = add_completed_job(
db,
completed_job,
false,
false,
result,
None,
mem_peak,
canceled_by,
flow_is_done,
duration,
false,
)
.warn_after_seconds(10)
.await?;
Ok(())
}
pub async fn add_completed_job_error(
db: &Pool<Postgres>,
completed_job: &MiniCompletedJob,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
worker_name: &str,
flow_is_done: bool,
duration: Option<i64>,
) -> Result<WrappedError, Error> {
record_failure_metrics(completed_job, worker_name).await;
let result = WrappedError { error: e };
tracing::error!(
backend/windmill-worker/src/result_processor.rs
+109 -26
View File
@@ -35,8 +35,9 @@ use windmill_common::{
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
use windmill_queue::{
append_logs, get_mini_completed_job, CanceledBy, FlowRunners, JobCompleted, MiniCompletedJob,
MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG,
append_logs, get_mini_completed_job, is_pre_shaped_wm_failure_result, CanceledBy, FlowRunners,
JobCompleted, MiniCompletedJob, MiniPulledJob, ValidableJson, WrappedError, INIT_SCRIPT_TAG,
MANUAL_FAILURE_ERROR_NAME,
};
use serde_json::{json, value::RawValue, Value};
@@ -61,8 +62,37 @@ struct ErrorMessage {
name: String,
}
#[derive(Debug, Deserialize)]
struct NestedErrorMessage {
error: ErrorMessage,
}
/// Extract `{ name, message }` from a result. Accepts both the standard
/// top-level shape (regular runtime errors) and the nested `{ error: { name,
/// message }, ... }` shape produced by the wm_failure injection.
///
/// For wm_failure-injected results, we prefer the nested error: a successful
/// run may legitimately contain top-level `name`/`message` fields (user data
/// named `name`/`message`), and we want OTel to record the ManualFailure
/// rather than the user's sibling fields.
fn extract_error_message(raw: &str) -> Option<ErrorMessage> {
let nested = serde_json::from_str::<NestedErrorMessage>(raw)
.ok()
.map(|n| n.error);
if matches!(&nested, Some(em) if em.name == MANUAL_FAILURE_ERROR_NAME) {
return nested;
}
if let Ok(em) = serde_json::from_str::<ErrorMessage>(raw) {
return Some(em);
}
nested
}
/// Returns the post-processing `success` value (after any `wm_failure`
/// override). Callers use this to make worker-loop decisions that depend on
/// whether the job ultimately succeeded — e.g. the init-script killpill.
async fn process_jc(
jc: JobCompleted,
mut jc: JobCompleted,
worker_name: &str,
base_internal_url: &str,
db: &DB,
@@ -73,7 +103,32 @@ async fn process_jc(
killpill_rx: &tokio::sync::broadcast::Receiver<()>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
#[cfg(feature = "benchmark")] bench_infos: &mut BenchmarkInfo,
) {
) -> bool {
// Parse `wm_labels` and `wm_failure` together (single `from_str`)
// so we don't deserialize the whole result twice on every job.
let metadata = jc.result.result_metadata();
// If the script returned a `wm_failure: <string>` field in its
// result, tag the run as a failure. Inject an `error: { name, message }`
// at the top level so error handlers / UI / OTel see the standard error
// shape, while preserving sibling fields (`windmill_status_code`,
// `windmill_content_type`, `windmill_headers`, the user's data) at the
// top level so sync webhook responses still honor them.
if jc.success {
if let Some(failure_msg) = metadata.wm_failure.as_ref() {
if let Ok(Value::Object(mut map)) = serde_json::from_str::<Value>(jc.result.get()) {
map.insert(
"error".to_string(),
json!({ "name": MANUAL_FAILURE_ERROR_NAME, "message": failure_msg }),
);
if let Ok(raw) = serde_json::value::to_raw_value(&Value::Object(map)) {
jc.result = Arc::new(raw);
}
}
jc.success = false;
}
}
let success: bool = jc.success;
let span = if success {
@@ -125,7 +180,7 @@ async fn process_jc(
jc.job.id
};
if let Some(labels) = jc.result.wm_labels() {
if let Some(labels) = metadata.wm_labels.as_ref() {
if !labels.is_empty() {
span.record("labels", labels.join(","));
}
@@ -163,7 +218,7 @@ async fn process_jc(
span.record("script_hash", script_hash.to_string().as_str());
}
if !success {
if let Ok(result_error) = serde_json::from_str::<ErrorMessage>(jc.result.get()) {
if let Some(result_error) = extract_error_message(jc.result.get()) {
span.record("error.message", result_error.message.as_str());
span.record("error.name", result_error.name.as_str());
span.record(
@@ -218,6 +273,8 @@ async fn process_jc(
)
.await;
}
success
}
enum JobCompletedRx {
@@ -311,8 +368,7 @@ pub fn start_background_processor(
result: SendResultPayload::JobCompleted(jc),
time,
}) => {
let is_init_script_and_failure =
!jc.success && jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_init_script = jc.job.tag.as_str() == INIT_SCRIPT_TAG;
let is_dependency_job = matches!(
jc.job.kind,
JobKind::Dependencies | JobKind::FlowDependencies
@@ -322,7 +378,10 @@ pub fn start_background_processor(
#[cfg(feature = "benchmark")]
let is_top_level_job = jc.job.parent_job.is_none();
process_jc(
// process_jc returns the post-override success value so a
// job that flipped to failure via `wm_failure` still
// triggers the init-script killpill.
let final_success = process_jc(
jc,
&worker_name,
&base_internal_url,
@@ -340,7 +399,7 @@ pub fn start_background_processor(
.warn_after_seconds(10)
.await;
if is_init_script_and_failure {
if is_init_script && !final_success {
tracing::error!("init script errored, exiting");
killpill_tx.send();
break;
@@ -793,19 +852,44 @@ pub async fn process_completed_job(
}
}
} else {
let result = add_completed_job_error(
db,
&job,
mem_peak.to_owned(),
canceled_by.clone(),
serde_json::from_str(result.get()).unwrap_or_else(
|_| json!({ "message": format!("Non serializable error: {}", result.get()) }),
),
worker_name,
false,
None,
)
.await?;
// The result already carries our injected
// `error: { name: "ManualFailure", ... }` marker when process_jc
// retagged a successful run as a failure — store it as-is to preserve
// sibling fields like `windmill_status_code`. We check for the
// injected marker specifically (not just the presence of a
// `wm_failure` field) so a real runtime failure whose raw
// result happens to contain a `wm_failure` field still goes
// through the standard `WrappedError { error: ... }` wrap path.
let downstream_result: Arc<Box<RawValue>> = if is_pre_shaped_wm_failure_result(result.get())
{
windmill_queue::add_completed_job_pre_shaped_failure(
db,
&job,
mem_peak.to_owned(),
canceled_by.clone(),
Json(&*result),
worker_name,
false,
None,
)
.await?;
result.clone()
} else {
let wrapped = add_completed_job_error(
db,
&job,
mem_peak.to_owned(),
canceled_by.clone(),
serde_json::from_str(result.get()).unwrap_or_else(
|_| json!({ "message": format!("Non serializable error: {}", result.get()) }),
),
worker_name,
false,
None,
)
.await?;
Arc::new(serde_json::value::to_raw_value(&wrapped).unwrap())
};
if job.is_flow_step() {
if let Some(parent_job) = job.parent_job {
tracing::error!(parent_flow = %parent_job, subflow = %job.id, "process completed job error, updating flow status");
@@ -817,7 +901,7 @@ pub async fn process_completed_job(
&job.workspace_id,
false,
canceled_by,
Arc::new(serde_json::value::to_raw_value(&result).unwrap()),
downstream_result,
duration.and_then(|d| {
job.started_at.map(|started_at| FlowJobDuration {
started_at: started_at,
@@ -855,13 +939,12 @@ pub async fn process_completed_job(
.fetch_optional(db)
.await?;
if let Some(Some(job_ids)) = job_ids_json {
let err_result = Arc::new(serde_json::value::to_raw_value(&result).unwrap());
if let Ok(Some(_)) = handle_wac_child_completion(
db,
&job.id,
parent_job,
&job.workspace_id,
err_result,
downstream_result,
false,
job_ids,
)