From d45ddecf8fdd14fe5ef759b7e836caacf1b78a52 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Fri, 1 Aug 2025 12:24:00 +0000
Subject: [PATCH] fix: add wm_labels to tracing spans

---
 .../windmill-worker/src/result_processor.rs   | 13 ++++++++---
 backend/windmill-worker/src/worker_flow.rs    | 23 ++++++-------------
 2 files changed, 17 insertions(+), 19 deletions(-)

diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs
index 86ea8f8eec..a5de05c9e9 100644
--- a/backend/windmill-worker/src/result_processor.rs
+++ b/backend/windmill-worker/src/result_processor.rs
@@ -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,
diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs
index 4250a23a97..00477e7510 100644
--- a/backend/windmill-worker/src/worker_flow.rs
+++ b/backend/windmill-worker/src/worker_flow.rs
@@ -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,