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

associate metrics with tags and not langs anymore

Browse Source
This commit is contained in:
Ruben Fiszel
2023-10-26 23:44:25 +02:00
parent 55ceca1913
commit dc10aaa660
5 changed files with 193 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/monitor.rs
+3 -2
View File
@@ -422,7 +422,7 @@ pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
) {
let zombie_jobs_f = async {
if server_mode {
handle_zombie_jobs(db, base_internal_url, rsmq.clone()).await;
handle_zombie_jobs(db, base_internal_url, rsmq.clone(), "server").await;
}
};
let expired_items_f = async {
@@ -608,6 +608,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
db: &Pool<Postgres>,
base_internal_url: &str,
rsmq: Option<R>,
worker_name: &str,
) {
if *RESTART_ZOMBIE_JOBS {
let restarted = sqlx::query!(
@@ -688,11 +689,11 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.unwrap_or_else(|| "no ping".to_string()),
*ZOMBIE_JOB_TIMEOUT
)),
None,
true,
same_worker_tx_never_used,
"",
rsmq.clone(),
worker_name,
)
.await;
}
backend/windmill-common/src/jobs.rs
-5
View File
@@ -330,8 +330,3 @@ pub async fn get_payload_tag_from_prefixed_path(
};
Ok((payload, tag))
}
#[derive(Clone)]
pub struct Metrics {
pub worker_execution_failed: prometheus::IntCounter,
}
backend/windmill-queue/src/jobs.rs
+60 -8
View File
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::{collections::HashMap, vec};
use std::{collections::HashMap, sync::Arc, vec};
use anyhow::Context;
use async_recursion::async_recursion;
@@ -19,6 +19,7 @@ use axum::{
};
use bigdecimal::ToPrimitive;
use chrono::{DateTime, Duration, Utc};
use prometheus::IntCounter;
use reqwest::{
header::{HeaderMap, CONTENT_TYPE},
Client, StatusCode,
@@ -29,6 +30,7 @@ use serde_json::{json, value::RawValue};
use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
#[cfg(feature = "benchmark")]
use std::time::Instant;
use tokio::sync::RwLock;
use tracing::{instrument, Instrument};
use ulid::Ulid;
use uuid::Uuid;
@@ -43,8 +45,8 @@ use windmill_common::{
},
flows::{FlowModule, FlowModuleValue, FlowValue},
jobs::{
get_payload_tag_from_prefixed_path, script_path_to_payload, JobKind, JobPayload, Metrics,
QueuedJob, RawCode,
get_payload_tag_from_prefixed_path, script_path_to_payload, JobKind, JobPayload, QueuedJob,
RawCode,
},
schedule::{schedule_to_user, Schedule},
scripts::{ScriptHash, ScriptLang},
@@ -72,17 +74,22 @@ lazy_static::lazy_static! {
"Total number of jobs pushed to the queue."
)
.unwrap();
static ref QUEUE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_delete_count",
"Total number of jobs deleted from the queue."
)
.unwrap();
static ref QUEUE_PULL_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_pull_count",
"Total number of jobs pulled from the queue."
)
.unwrap();
pub static ref WORKER_EXECUTION_FAILED: Arc<RwLock<HashMap<String, IntCounter>>> = Arc::new(RwLock::new(HashMap::new()));
}
#[cfg(feature = "enterprise")]
@@ -135,8 +142,8 @@ pub async fn cancel_job<'c: 'async_recursion>(
format!("canceled by {username}: (force cancel: {force_cancel})"),
job_running.mem_peak.unwrap_or(0),
e,
None,
rsmq.clone(),
"server",
)
.await;
if let Err(e) = add_job {
@@ -185,6 +192,36 @@ pub struct WrappedError {
pub error: serde_json::Value,
}
pub async fn register_metric<T, F, F2, R>(
l: &Arc<RwLock<HashMap<String, T>>>,
s: &str,
fnew: F2,
f: F,
) -> Option<R>
where
F: FnOnce(&T) -> R,
F2: FnOnce(&str) -> (T, R),
{
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
let lock = l.read().await;
let counter = lock.get(s);
if let Some(counter) = counter {
let r = f(counter);
drop(lock);
Some(r)
} else {
drop(lock);
let (metric, r) = fnew(s);
let mut m = l.write().await;
(*m).insert(s.to_string(), metric);
Some(r)
}
} else {
None
}
}
#[instrument(level = "trace", skip_all)]
pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Send>(
db: &Pool<Postgres>,
@@ -192,12 +229,27 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
logs: String,
mem_peak: i32,
e: serde_json::Value,
metrics: Option<Metrics>,
rsmq: Option<R>,
worker_name: &str,
) -> Result<WrappedError, Error> {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
metrics.map(|m| m.worker_execution_failed.inc());
}
register_metric(
&WORKER_EXECUTION_FAILED,
&queued_job.tag,
|s| {
let counter = prometheus::register_int_counter!(prometheus::Opts::new(
"worker_execution_count",
"Number of executed jobs"
)
.const_label("name", worker_name)
.const_label("tag", s))
.expect("register prometheus metric");
counter.inc();
(counter, ())
},
|c| c.inc(),
)
.await;
let result = WrappedError { error: e };
tracing::error!(
"job {} did not succeed: {}",
backend/windmill-worker/src/worker.rs
+115 -140
View File
@@ -9,7 +9,7 @@
use anyhow::Result;
use const_format::concatcp;
use itertools::Itertools;
use prometheus::core::{AtomicU64, GenericCounter};
use prometheus::IntCounter;
use reqwest::Response;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use sqlx::{types::Json, Pool, Postgres};
@@ -26,7 +26,7 @@ use uuid::Uuid;
use windmill_common::{
error::{self, to_anyhow, Error},
flows::{FlowModule, FlowModuleValue, FlowValue},
jobs::{JobKind, Metrics, QueuedJob},
jobs::{JobKind, QueuedJob},
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
users::SUPERADMIN_SECRET_EMAIL,
utils::{rd_string, StripPath},
@@ -36,8 +36,8 @@ use windmill_common::{
DB, IS_READY, METRICS_ENABLED,
};
use windmill_queue::{
canceled_job_to_result, empty_args, get_queued_job, pull, push, PushArgs, PushIsolationLevel,
WrappedError, HTTP_CLIENT,
canceled_job_to_result, empty_args, get_queued_job, pull, push, register_metric, PushArgs,
PushIsolationLevel, WrappedError, HTTP_CLIENT,
};
use serde_json::{json, value::RawValue, Value};
@@ -283,6 +283,10 @@ lazy_static::lazy_static! {
pub static ref CAN_PULL: Arc<RwLock<()>> = Arc::new(RwLock::new(()));
pub static ref WORKER_EXECUTION_COUNT: Arc<RwLock<HashMap<String, IntCounter>>> = Arc::new(RwLock::new(HashMap::new()));
pub static ref WORKER_EXECUTION_DURATION_COUNTER: Arc<RwLock<HashMap<String, prometheus::Counter>>> = Arc::new(RwLock::new(HashMap::new()));
pub static ref WORKER_EXECUTION_DURATION: Arc<RwLock<HashMap<String, prometheus::Histogram>>> = Arc::new(RwLock::new(HashMap::new()));
}
//only matter if CLOUD_HOSTED
@@ -445,14 +449,13 @@ async fn handle_receive_completed_job<
R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static,
>(
jc: JobCompleted,
worker_execution_failed: HashMap<Option<ScriptLang>, GenericCounter<AtomicU64>>,
base_internal_url: String,
db: Pool<Postgres>,
worker_dir: String,
same_worker_tx: Sender<Uuid>,
rsmq: Option<R>,
worker_name: &str,
) {
let metrics = build_language_metrics(&worker_execution_failed.clone(), &jc.job.language);
let token = jc.token.clone();
let workspace = jc.job.workspace_id.clone();
let client = AuthedClient {
@@ -468,9 +471,9 @@ async fn handle_receive_completed_job<
&client,
&db,
&worker_dir,
metrics.clone(),
same_worker_tx.clone(),
rsmq.clone(),
worker_name,
)
.await
{
@@ -480,11 +483,11 @@ async fn handle_receive_completed_job<
job.as_ref(),
mem_peak,
err,
metrics,
false,
same_worker_tx.clone(),
&worker_dir,
rsmq.clone(),
worker_name,
)
.await;
}
@@ -543,55 +546,22 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
prometheus::register_counter!(WORKER_UPTIME_OPTS.clone().const_label("name", &worker_name))
.unwrap();
let all_langs = [
None,
Some(ScriptLang::Python3),
Some(ScriptLang::Deno),
Some(ScriptLang::Go),
Some(ScriptLang::Bash),
Some(ScriptLang::Powershell),
Some(ScriptLang::Nativets),
Some(ScriptLang::Postgresql),
Some(ScriptLang::Mysql),
Some(ScriptLang::Bigquery),
Some(ScriptLang::Snowflake),
Some(ScriptLang::Graphql),
Some(ScriptLang::Bun),
];
let worker_execution_duration: HashMap<_, _> = all_langs
.clone()
.into_iter()
.map(|x| {
(
x.clone(),
prometheus::register_histogram!(prometheus::HistogramOpts::new(
"worker_execution_duration",
"Duration between receiving a job and completing it",
)
.const_label("name", &worker_name)
.const_label("language", x.map(|x| x.as_str()).unwrap_or("none")))
.expect("register prometheus metric"),
)
})
.collect();
let worker_execution_duration_counter: HashMap<_, _> = all_langs
.clone()
.into_iter()
.map(|x| {
(
x.clone(),
prometheus::register_counter!(prometheus::Opts::new(
"worker_execution_duration_counter",
"Total number of seconds spent executing jobs"
)
.const_label("name", &worker_name)
.const_label("language", x.map(|x| x.as_str()).unwrap_or("none")))
.expect("register prometheus metric"),
)
})
.collect();
// let worker_execution_duration_counter: HashMap<_, _> = all_langs
// .clone()
// .into_iter()
// .map(|x| {
// (
// x.clone(),
// prometheus::register_counter!(prometheus::Opts::new(
// "worker_execution_duration_counter",
// "Total number of seconds spent executing jobs"
// )
// .const_label("name", &worker_name)
// .const_label("language", x.map(|x| x.as_str()).unwrap_or("none")))
// .expect("register prometheus metric"),
// )
// })
// .collect();
let worker_sleep_duration_counter = prometheus::register_counter!(prometheus::opts!(
"worker_sleep_duration_counter",
@@ -614,39 +584,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.const_label("name", &worker_name))
.expect("register prometheus metric");
let worker_execution_failed: HashMap<_, _> = all_langs
.clone()
.into_iter()
.map(|x| {
(
x.clone(),
prometheus::register_int_counter!(prometheus::Opts::new(
"worker_execution_failed",
"Number of failed jobs",
)
.const_label("name", &worker_name)
.const_label("language", x.map(|x| x.as_str()).unwrap_or("none")))
.expect("register prometheus metric"),
)
})
.collect();
let worker_execution_count: HashMap<_, _> = all_langs
.into_iter()
.map(|x| {
(
x.clone(),
prometheus::register_int_counter!(prometheus::Opts::new(
"worker_execution_count",
"Number of executed jobs"
)
.const_label("name", &worker_name)
.const_label("language", x.map(|x| x.as_str()).unwrap_or("none")))
.expect("register prometheus metric"),
)
})
.collect();
let worker_busy: prometheus::IntGauge = prometheus::register_int_gauge!(prometheus::Opts::new(
"worker_busy",
"Is the worker busy executing a job?",
@@ -699,7 +636,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let same_worker_tx2 = same_worker_tx.clone();
let rsmq2 = rsmq.clone();
let worker_dir2 = worker_dir.clone();
let worker_execution_failed2 = worker_execution_failed.clone();
let thread_count = Arc::new(AtomicUsize::new(0));
let is_dedicated_worker = WORKER_CONFIG.read().await.dedicated_worker.is_some();
@@ -769,14 +705,15 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
#[cfg(feature = "benchmark")]
let send_duration2 = send_duration.clone();
let worker_name2 = worker_name.clone();
let send_result = tokio::spawn(async move {
while let Some(jc) = job_completed_rx.recv().await {
let base_internal_url2 = base_internal_url2.clone();
let worker_execution_failed2 = worker_execution_failed2.clone();
let worker_dir2 = worker_dir2.clone();
let db2 = db2.clone();
let same_worker_tx2 = same_worker_tx2.clone();
let rsmq2 = rsmq2.clone();
let worker_name = worker_name2.clone();
if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker {
thread_count.fetch_add(1, Ordering::SeqCst);
@@ -797,12 +734,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
handle_receive_completed_job(
jc,
worker_execution_failed2,
base_internal_url2,
db2,
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
)
.await;
#[cfg(feature = "benchmark")]
@@ -845,12 +782,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
} else {
handle_receive_completed_job(
jc,
worker_execution_failed2,
base_internal_url2,
db2,
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
)
.await;
}
@@ -1238,18 +1175,42 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
} else {
let token = create_token_for_owner_in_bg(&db, &job).await;
let language = job.language.clone();
let _timer = worker_execution_duration
.get(&language)
.expect("no timer found")
.start_timer();
register_metric(
&WORKER_EXECUTION_COUNT,
&job.tag,
|s| {
let counter = prometheus::register_int_counter!(prometheus::Opts::new(
"worker_execution_count",
"Number of executed jobs"
)
.const_label("name", &worker_name)
.const_label("tag", s))
.expect("register prometheus metric");
counter.inc();
(counter, ())
},
|c| c.inc(),
)
.await;
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
worker_execution_count
.get(&language)
.expect("no timer found")
.inc();
}
let _timer = register_metric(
&WORKER_EXECUTION_DURATION,
&job.tag,
|s| {
let counter =
prometheus::register_histogram!(prometheus::HistogramOpts::new(
"worker_execution_duration",
"Duration between receiving a job and completing it",
)
.const_label("name", &worker_name)
.const_label("tag", s))
.expect("register prometheus metric");
let t = counter.start_timer();
(counter, t)
},
|c| c.start_timer(),
)
.await;
let job_root = job
.root_job
@@ -1300,6 +1261,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
workspace: job.workspace_id.to_string(),
};
let tag = job.tag.clone();
let arc_job = Arc::new(job);
if let Some(err) = handle_queued_job(
arc_job.clone(),
@@ -1316,28 +1278,40 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.await
.err()
{
let metrics = build_language_metrics(&worker_execution_failed, &language);
handle_job_error(
db,
&authed_client.get_authed().await,
arc_job.as_ref(),
0,
err,
metrics,
false,
same_worker_tx.clone(),
&worker_dir,
rsmq.clone(),
&worker_name,
)
.await;
};
let duration = _timer.stop_and_record();
worker_execution_duration_counter
.get(&language)
.expect("no timer found")
.inc_by(duration);
if let Some(duration) = _timer.map(|x| x.stop_and_record()) {
register_metric(
&WORKER_EXECUTION_DURATION_COUNTER,
&tag,
|s| {
let counter = prometheus::register_counter!(prometheus::Opts::new(
"worker_execution_duration_counter",
"Total number of seconds spent executing jobs"
)
.const_label("name", &worker_name)
.const_label("tag", s))
.expect("register prometheus metric");
counter.inc_by(duration);
(counter, ())
},
|c| c.inc_by(duration),
)
.await;
}
if !KEEP_JOB_DIR.load(Ordering::Relaxed) && !(arc_job.is_flow() && same_worker)
{
@@ -1477,9 +1451,9 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
client: &AuthedClient,
db: &DB,
worker_dir: &str,
metrics: Option<Metrics>,
same_worker_tx: Sender<Uuid>,
rsmq: Option<R>,
worker_name: &str,
) -> windmill_common::error::Result<()> {
if success {
// println!("bef completed job{:?}", SystemTime::now());
@@ -1507,12 +1481,12 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
&job.workspace_id,
true,
&result,
metrics.clone(),
false,
same_worker_tx.clone(),
&worker_dir,
None,
rsmq.clone(),
worker_name,
)
.await?;
}
@@ -1526,8 +1500,8 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
serde_json::from_str(result.get()).unwrap_or_else(
|_| json!({ "message": format!("Non serializable error: {}", result.get()) }),
),
metrics.clone(),
rsmq.clone(),
worker_name,
)
.await?;
if job.is_flow_step {
@@ -1540,12 +1514,12 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
&job.workspace_id,
false,
&serde_json::value::to_raw_value(&result).unwrap(),
metrics,
false,
same_worker_tx,
&worker_dir,
None,
rsmq,
worker_name,
)
.await?;
}
@@ -1554,25 +1528,25 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
Ok(())
}
fn build_language_metrics(
worker_execution_failed: &HashMap<
Option<ScriptLang>,
prometheus::core::GenericCounter<prometheus::core::AtomicU64>,
>,
language: &Option<ScriptLang>,
) -> Option<Metrics> {
let metrics = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
Some(Metrics {
worker_execution_failed: worker_execution_failed
.get(language)
.expect("no timer found")
.clone(),
})
} else {
None
};
metrics
}
// fn build_language_metrics(
// worker_execution_failed: &HashMap<
// Option<ScriptLang>,
// prometheus::core::GenericCounter<prometheus::core::AtomicU64>,
// >,
// language: &Option<ScriptLang>,
// ) -> Option<Metrics> {
// let metrics = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
// Some(Metrics {
// worker_execution_failed: worker_execution_failed
// .get(language)
// .expect("no timer found")
// .clone(),
// })
// } else {
// None
// };
// metrics
// }
// pub async fn create_barrier_for_all_workers(num_workers: u32, sync_barrier: Arc<RwLock<Option<tokio::sync::Barrier>>>) {
// tracing::debug!("acquiring write lock");
@@ -1595,11 +1569,11 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
job: &QueuedJob,
mem_peak: i32,
err: Error,
metrics: Option<Metrics>,
unrecoverable: bool,
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
) {
let err = match err {
Error::JsonErr(err) => err,
@@ -1614,8 +1588,8 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
format!("Unexpected error during job execution:\n{err:#?}"),
mem_peak,
err.clone(),
metrics.clone(),
rsmq_2,
worker_name,
)
};
@@ -1640,12 +1614,12 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
&job.workspace_id,
false,
&serde_json::value::to_raw_value(&wrapped_error).unwrap(),
metrics.clone(),
unrecoverable,
same_worker_tx,
worker_dir,
None,
rsmq.clone(),
worker_name,
)
.await;
@@ -1662,8 +1636,8 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
e,
metrics.clone(),
rsmq,
worker_name,
)
.await;
}
@@ -1849,6 +1823,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx,
worker_dir,
rsmq,
worker_name,
)
.await?;
}
backend/windmill-worker/src/worker_flow.rs
+15 -11
View File
@@ -7,8 +7,8 @@
*/
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use crate::common::{hash_args, save_in_cache};
@@ -28,7 +28,7 @@ use windmill_common::flow_status::{
ApprovalConditions, FlowStatusModuleWParent, Iterator, JobResult,
};
use windmill_common::jobs::{
script_hash_to_tag_and_limits, script_path_to_payload, JobPayload, Metrics, QueuedJob, RawCode,
script_hash_to_tag_and_limits, script_path_to_payload, JobPayload, QueuedJob, RawCode,
};
use windmill_common::worker::to_raw_value;
use windmill_common::{
@@ -60,12 +60,12 @@ pub async fn update_flow_status_after_job_completion<
w_id: &str,
success: bool,
result: &'a RawValue,
metrics: Option<Metrics>,
unrecoverable: bool,
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
stop_early_override: Option<bool>,
rsmq: Option<R>,
worker_name: &str,
) -> error::Result<()> {
// this is manual tailrecursion because async_recursion blows up the stack
let mut rec = update_flow_status_after_job_completion_internal(
@@ -76,13 +76,13 @@ pub async fn update_flow_status_after_job_completion<
w_id,
success,
result,
metrics.clone(),
unrecoverable,
same_worker_tx.clone(),
worker_dir,
stop_early_override,
false,
rsmq.clone(),
worker_name,
)
.await?;
while let Some(nrec) = rec {
@@ -94,13 +94,13 @@ pub async fn update_flow_status_after_job_completion<
w_id,
nrec.success,
nrec.result.as_ref(),
metrics.clone(),
false,
same_worker_tx.clone(),
worker_dir,
nrec.stop_early_override,
nrec.skip_error_handler,
rsmq.clone(),
worker_name,
)
.await
{
@@ -114,13 +114,13 @@ pub async fn update_flow_status_after_job_completion<
w_id,
false,
&to_raw_value(&Json(&WrappedError { error: json!(e.to_string()) })),
metrics.clone(),
true,
same_worker_tx.clone(),
worker_dir,
nrec.stop_early_override,
nrec.skip_error_handler,
rsmq.clone(),
worker_name,
)
.await?
}
@@ -162,13 +162,13 @@ pub async fn update_flow_status_after_job_completion_internal<
w_id: &str,
mut success: bool,
result: &'a RawValue,
metrics: Option<Metrics>,
unrecoverable: bool,
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
stop_early_override: Option<bool>,
skip_error_handler: bool,
rsmq: Option<R>,
worker_name: &str,
) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
let (should_continue_flow, flow_job, stop_early, skip_if_stop_early, nresult, is_failure_step) = {
// tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id} {depth}");
@@ -639,8 +639,8 @@ pub async fn update_flow_status_after_job_completion_internal<
logs,
0,
canceled_job_to_result(&flow_job),
metrics.clone(),
rsmq.clone(),
worker_name,
)
.await?;
} else {
@@ -694,6 +694,7 @@ pub async fn update_flow_status_after_job_completion_internal<
same_worker_tx.clone(),
worker_dir,
rsmq.clone(),
worker_name,
)
.await
{
@@ -705,8 +706,8 @@ pub async fn update_flow_status_after_job_completion_internal<
"Unexpected error during flow chaining:\n".to_string(),
0,
e,
metrics.clone(),
rsmq.clone(),
worker_name,
)
.await;
true
@@ -987,6 +988,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
) -> anyhow::Result<()> {
let value = flow_job
.raw_flow
@@ -1011,6 +1013,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx,
worker_dir,
rsmq,
worker_name,
)
.await?;
Ok(())
@@ -1055,6 +1058,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
) -> error::Result<()> {
let job_root = flow_job
.root_job
@@ -1090,12 +1094,12 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
// it has to be an empty for loop event
serde_json::from_str("[]").unwrap()
},
None,
true,
same_worker_tx,
worker_dir,
None,
rsmq,
worker_name,
)
.await;
}
@@ -1123,12 +1127,12 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
flow_job.workspace_id.as_str(),
true,
serde_json::from_str("\"stopped early\"").unwrap(),
None,
true,
same_worker_tx,
worker_dir,
Some(true),
rsmq,
worker_name,
)
.await;
}