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

fix: add wm_labels to tracing spans

Browse Source
This commit is contained in:
Ruben Fiszel
2025-08-01 12:24:00 +00:00
parent c279154409
commit d45ddecf8f
2 changed files with 17 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-worker/src/result_processor.rs
+10 -3
View File
@@ -29,8 +29,8 @@ use windmill_common::{
use windmill_common::bench::{BenchmarkInfo, BenchmarkIter};
use windmill_queue::{
append_logs, get_queued_job, CanceledBy, JobCompleted, MiniPulledJob, WrappedError,
INIT_SCRIPT_TAG,
append_logs, get_queued_job, CanceledBy, JobCompleted, MiniPulledJob, ValidableJson,
WrappedError, INIT_SCRIPT_TAG,
};
use serde_json::{json, value::RawValue, Value};
@@ -79,6 +79,7 @@ async fn process_jc(
parent_job = field::Empty,
otel.name = field::Empty,
success = %success,
labels = field::Empty,
)
} else {
tracing::span!(
@@ -94,6 +95,7 @@ async fn process_jc(
success = %success,
error.message = field::Empty,
error.name = field::Empty,
labels = field::Empty,
)
};
let rj = if let Some(root_job) = jc.job.flow_innermost_root_job {
@@ -101,6 +103,12 @@ async fn process_jc(
} else {
jc.job.id
};
if let Some(labels) = jc.result.wm_labels() {
if !labels.is_empty() {
span.record("labels", labels.join(","));
}
}
windmill_common::otel_oss::set_span_parent(&span, &rj);
if let Some(lg) = jc.job.script_lang.as_ref() {
@@ -686,7 +694,6 @@ pub async fn handle_job_error(
let err_string = format!("{}: {}", err.name(), err.to_string());
let err_json = error_to_value(err);
let update_job_future = || async {
handle_non_flow_job_error(
db,
backend/windmill-worker/src/worker_flow.rs
+7 -16
View File
@@ -575,7 +575,6 @@ pub async fn update_flow_status_after_job_completion_internal(
}
}
let branches = current_module
.and_then(|x| x.get_branches_skip_failures().ok())
.map(|x| {
@@ -601,8 +600,6 @@ pub async fn update_flow_status_after_job_completion_internal(
} else {
jobs.as_slice()
};
let new_status = if skip_loop_failures
|| sqlx::query_scalar!(
@@ -838,9 +835,6 @@ pub async fn update_flow_status_after_job_completion_internal(
}
};
let step_counter = if inc_step_counter {
sqlx::query!(
"UPDATE v2_job_status
@@ -1152,18 +1146,12 @@ pub async fn update_flow_status_after_job_completion_internal(
.map(|x| x.to_string())
.unwrap_or_else(|| "none".to_string());
let should_continue_flow = match success {
_ if stop_early => false,
_ if flow_job.is_canceled() => false,
true => !is_last_step,
false if unrecoverable => false,
false
if skip_seq_branch_failure
|| skip_loop_failures
|| continue_on_error =>
{
false if skip_seq_branch_failure || skip_loop_failures || continue_on_error => {
!is_last_step
}
false
@@ -1227,9 +1215,12 @@ pub async fn update_flow_status_after_job_completion_internal(
} else if stop_early {
format!("Flow job stopped early because of a stop early predicate returning true\n")
} else if is_failure_step {
format!("Flow job completed with error, and error handler was triggered.\n It completed with {}, and with recover: {}\n", if success { "success" } else { "error" }, result_has_recover_true(nresult.clone()))
format!("Flow job completed with error, and error handler was triggered.\nIt completed with {}, and with recover: {}\n", if success { "success" } else { "error" }, result_has_recover_true(nresult.clone()))
} else {
format!("Flow job completed with {}\n", if success { "success" } else { "error" })
format!(
"Flow job completed with {}\n",
if success { "success" } else { "error" }
)
};
append_logs(&flow_job.id, w_id, logs, &db.into()).await;
}
@@ -1279,7 +1270,7 @@ pub async fn update_flow_status_after_job_completion_internal(
})?;
}
}
if flow_job.is_canceled() {
add_completed_job_error(
db,