Compare commits
1 Commits
wmill-scri
...
uael/more_
| Author | SHA1 | Date | |
|---|---|---|---|
|
d823c546f2 |
@@ -4,10 +4,7 @@ use std::{
|
||||
fmt::Display,
|
||||
ops::Mul,
|
||||
str::FromStr,
|
||||
sync::{
|
||||
atomic::{AtomicU16, Ordering},
|
||||
Arc,
|
||||
},
|
||||
sync::{atomic::Ordering, Arc},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
@@ -15,11 +12,7 @@ use chrono::{NaiveDateTime, Utc};
|
||||
use futures::{stream::FuturesUnordered, StreamExt};
|
||||
use serde::{de::DeserializeOwned, Deserializer};
|
||||
use sqlx::{Pool, Postgres};
|
||||
use tokio::{
|
||||
join,
|
||||
sync::{mpsc, RwLock},
|
||||
};
|
||||
use uuid::Uuid;
|
||||
use tokio::{join, sync::RwLock};
|
||||
|
||||
#[cfg(feature = "embedding")]
|
||||
use windmill_api::embeddings::update_embeddings_db;
|
||||
@@ -37,7 +30,6 @@ use windmill_common::{
|
||||
auth::JWT_SECRET,
|
||||
ee::CriticalErrorChannel,
|
||||
error,
|
||||
flow_status::{FlowStatus, FlowStatusModule},
|
||||
global_settings::{
|
||||
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
|
||||
CRITICAL_ERROR_CHANNELS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
|
||||
@@ -50,12 +42,11 @@ use windmill_common::{
|
||||
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
|
||||
},
|
||||
indexer::load_indexer_config,
|
||||
jobs::QueuedJob,
|
||||
oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
|
||||
server::load_smtp_config,
|
||||
tracing_init::JSON_FMT,
|
||||
users::truncate_token,
|
||||
utils::{now_from_db, rd_string, report_critical_error, Mode},
|
||||
utils::{rd_string, Mode},
|
||||
worker::{
|
||||
load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting,
|
||||
update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, INDEXER_CONFIG,
|
||||
@@ -66,11 +57,9 @@ use windmill_common::{
|
||||
MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED,
|
||||
SERVICE_LOG_RETENTION_SECS,
|
||||
};
|
||||
use windmill_queue::cancel_job;
|
||||
use windmill_worker::{
|
||||
create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender,
|
||||
SendResult, BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR,
|
||||
NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, SCRIPT_TOKEN_EXPIRY,
|
||||
BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR,
|
||||
NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL,
|
||||
};
|
||||
|
||||
#[cfg(feature = "parquet")]
|
||||
@@ -89,18 +78,6 @@ use crate::ee::set_license_key;
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
lazy_static::lazy_static! {
|
||||
|
||||
static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
|
||||
"queue_zombie_restart_count",
|
||||
"Total number of jobs restarted due to ping timeout."
|
||||
)
|
||||
.unwrap();
|
||||
static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
|
||||
"queue_zombie_delete_count",
|
||||
"Total number of jobs deleted due to their ping timing out in an unrecoverable state."
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
static ref QUEUE_COUNT: prometheus::IntGaugeVec = prometheus::register_int_gauge_vec!(
|
||||
"queue_count",
|
||||
"Number of jobs in the queue",
|
||||
@@ -109,26 +86,7 @@ lazy_static::lazy_static! {
|
||||
|
||||
}
|
||||
lazy_static::lazy_static! {
|
||||
static ref ZOMBIE_JOB_TIMEOUT: String = std::env::var("ZOMBIE_JOB_TIMEOUT")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<String>().ok())
|
||||
.unwrap_or_else(|| "60".to_string());
|
||||
|
||||
static ref FLOW_ZOMBIE_TRANSITION_TIMEOUT: String = std::env::var("FLOW_ZOMBIE_TRANSITION_TIMEOUT")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<String>().ok())
|
||||
.unwrap_or_else(|| "60".to_string());
|
||||
|
||||
|
||||
pub static ref RESTART_ZOMBIE_JOBS: bool = std::env::var("RESTART_ZOMBIE_JOBS")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<bool>().ok())
|
||||
.unwrap_or(true);
|
||||
|
||||
|
||||
|
||||
static ref QUEUE_COUNT_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(Vec::new()));
|
||||
|
||||
}
|
||||
|
||||
pub async fn initial_load(
|
||||
@@ -1232,13 +1190,15 @@ pub async fn monitor_db(
|
||||
) {
|
||||
let zombie_jobs_f = async {
|
||||
if server_mode && !initial_load {
|
||||
handle_zombie_jobs(db, base_internal_url, "server").await;
|
||||
match handle_zombie_flows(db).await {
|
||||
Err(err) => {
|
||||
tracing::error!("Error handling zombie flows: {:?}", err);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
windmill_worker::monitor::monitor_once(
|
||||
&db,
|
||||
base_internal_url,
|
||||
"server",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
};
|
||||
let expired_items_f = async {
|
||||
@@ -1517,346 +1477,6 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str, worker_name: &str) {
|
||||
if *RESTART_ZOMBIE_JOBS {
|
||||
let restarted = sqlx::query!(
|
||||
"WITH zombie_jobs AS (
|
||||
UPDATE v2_job_queue q SET running = false, started_at = null
|
||||
FROM v2_job j, v2_job_runtime r
|
||||
WHERE j.id = q.id AND j.id = r.id
|
||||
AND ping < now() - ($1 || ' seconds')::interval
|
||||
AND running = true
|
||||
AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')
|
||||
AND same_worker = false
|
||||
RETURNING q.id, q.workspace_id, ping
|
||||
),
|
||||
update_concurrency AS (
|
||||
UPDATE concurrency_counter cc
|
||||
SET job_uuids = job_uuids - zj.id::text
|
||||
FROM zombie_jobs zj
|
||||
INNER JOIN concurrency_key ck ON ck.job_id = zj.id
|
||||
WHERE cc.concurrency_id = ck.key
|
||||
)
|
||||
SELECT id, workspace_id, ping FROM zombie_jobs",
|
||||
*ZOMBIE_JOB_TIMEOUT,
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![]);
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
|
||||
}
|
||||
|
||||
let base_url = BASE_URL.read().await.clone();
|
||||
for r in restarted {
|
||||
let last_ping = if let Some(x) = r.ping {
|
||||
format!("last ping at {x}")
|
||||
} else {
|
||||
"no last ping".to_string()
|
||||
};
|
||||
let url = format!("{}/run/{}?workspace={}", base_url, r.id, r.workspace_id,);
|
||||
let error_message = format!(
|
||||
"Zombie job {} on {} ({}) detected, restarting it, {}",
|
||||
r.id, r.workspace_id, url, last_ping
|
||||
);
|
||||
|
||||
let _ = sqlx::query!("
|
||||
INSERT INTO job_logs (job_id, logs)
|
||||
VALUES ($1, 'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n')
|
||||
ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\n' || EXCLUDED.logs
|
||||
WHERE job_logs.job_id = $1",
|
||||
r.id
|
||||
)
|
||||
.execute(db)
|
||||
.await;
|
||||
tracing::error!(error_message);
|
||||
report_critical_error(error_message, db.clone(), Some(&r.workspace_id), None).await;
|
||||
}
|
||||
}
|
||||
|
||||
let same_worker_timeout_jobs = {
|
||||
let long_same_worker_jobs = sqlx::query!(
|
||||
"SELECT worker, array_agg(v2_job_queue.id) as ids FROM v2_job_queue LEFT JOIN v2_job ON v2_job_queue.id = v2_job.id LEFT JOIN v2_job_runtime ON v2_job_queue.id = v2_job_runtime.id WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval
|
||||
AND running = true AND ping IS NULL AND same_worker = true AND worker IS NOT NULL GROUP BY worker",
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![]);
|
||||
|
||||
let worker_ids = long_same_worker_jobs
|
||||
.iter()
|
||||
.map(|x| x.worker.clone().unwrap_or_default())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let long_dead_workers: std::collections::HashSet<String> = sqlx::query_scalar!(
|
||||
"WITH worker_ids AS (SELECT unnest($1::text[]) as worker)
|
||||
SELECT worker_ids.worker FROM worker_ids
|
||||
LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker
|
||||
WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval",
|
||||
&worker_ids[..]
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![])
|
||||
.into_iter()
|
||||
.filter_map(|x| x)
|
||||
.collect();
|
||||
|
||||
let mut timeouts: Vec<Uuid> = vec![];
|
||||
for worker in long_same_worker_jobs {
|
||||
if worker.worker.is_some() && long_dead_workers.contains(&worker.worker.unwrap()) {
|
||||
if let Some(ids) = worker.ids {
|
||||
timeouts.extend(ids);
|
||||
}
|
||||
}
|
||||
}
|
||||
if !timeouts.is_empty() {
|
||||
tracing::error!(
|
||||
"Failing same worker zombie jobs: {:?}",
|
||||
timeouts
|
||||
.iter()
|
||||
.map(|x| x.hyphenated().to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
);
|
||||
}
|
||||
|
||||
let jobs = sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)")
|
||||
.bind(&timeouts[..])
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e))
|
||||
.unwrap_or_default();
|
||||
|
||||
jobs
|
||||
};
|
||||
|
||||
let non_restartable_jobs = if *RESTART_ZOMBIE_JOBS {
|
||||
vec![]
|
||||
} else {
|
||||
sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE last_ping < now() - ($1 || ' seconds')::interval
|
||||
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow') AND same_worker = false")
|
||||
.bind(ZOMBIE_JOB_TIMEOUT.as_str())
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![])
|
||||
};
|
||||
|
||||
let timeouts = non_restartable_jobs
|
||||
.into_iter()
|
||||
.chain(same_worker_timeout_jobs)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
|
||||
}
|
||||
|
||||
for job in timeouts {
|
||||
// since the job is unrecoverable, the same worker queue should never be sent anything
|
||||
let (same_worker_tx_never_used, _same_worker_rx_never_used) =
|
||||
mpsc::channel::<SameWorkerPayload>(1);
|
||||
let same_worker_tx_never_used =
|
||||
SameWorkerSender(same_worker_tx_never_used, Arc::new(AtomicU16::new(0)));
|
||||
let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::<SendResult>(1);
|
||||
|
||||
let label = if job.permissioned_as != format!("u/{}", job.created_by)
|
||||
&& job.permissioned_as != job.created_by
|
||||
{
|
||||
format!("ephemeral-script-end-user-{}", job.created_by)
|
||||
} else {
|
||||
"ephemeral-script".to_string()
|
||||
};
|
||||
let token = create_token_for_owner(
|
||||
&db,
|
||||
&job.workspace_id,
|
||||
&job.permissioned_as,
|
||||
&label,
|
||||
*SCRIPT_TOKEN_EXPIRY,
|
||||
&job.email,
|
||||
&job.id,
|
||||
)
|
||||
.await
|
||||
.expect("could not create job token");
|
||||
|
||||
let client = AuthedClient {
|
||||
base_internal_url: base_internal_url.to_string(),
|
||||
token,
|
||||
workspace: job.workspace_id.to_string(),
|
||||
force_client: None,
|
||||
};
|
||||
|
||||
let last_ping = job.last_ping.clone();
|
||||
let _ = handle_job_error(
|
||||
db,
|
||||
&client,
|
||||
&job,
|
||||
0,
|
||||
None,
|
||||
error::Error::ExecutionErr(format!(
|
||||
"Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {}, same_worker: {})",
|
||||
last_ping
|
||||
.map(|x| x.to_string())
|
||||
.unwrap_or_else(|| "no ping".to_string()),
|
||||
*ZOMBIE_JOB_TIMEOUT,
|
||||
job.same_worker
|
||||
)),
|
||||
true,
|
||||
same_worker_tx_never_used,
|
||||
"",
|
||||
worker_name,
|
||||
send_result_never_used,
|
||||
#[cfg(feature = "benchmark")]
|
||||
&mut windmill_worker::bench::BenchmarkIter::new(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_zombie_flows(db: &DB) -> error::Result<()> {
|
||||
let flows = sqlx::query!(
|
||||
r#"
|
||||
SELECT
|
||||
id AS "id!", workspace_id AS "workspace_id!", parent_job, is_flow_step,
|
||||
flow_status AS "flow_status: Box<str>", last_ping, same_worker
|
||||
FROM v2_as_queue
|
||||
WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now()
|
||||
AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode')
|
||||
AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
|
||||
AND canceled = false
|
||||
"#,
|
||||
FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
for flow in flows {
|
||||
let status = flow
|
||||
.flow_status
|
||||
.as_deref()
|
||||
.and_then(|x| serde_json::from_str::<FlowStatus>(x).ok());
|
||||
if !flow.same_worker.unwrap_or(false)
|
||||
&& status.is_some_and(|s| {
|
||||
s.modules
|
||||
.get(0)
|
||||
.is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))
|
||||
})
|
||||
{
|
||||
let error_message = format!(
|
||||
"Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
|
||||
flow.id, flow.workspace_id
|
||||
);
|
||||
tracing::error!(error_message);
|
||||
report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await;
|
||||
// if the flow hasn't started and is a zombie, we can simply restart it
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
let concurrency_key =
|
||||
sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
|
||||
if let Some(key) = concurrency_key {
|
||||
sqlx::query!(
|
||||
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
|
||||
key,
|
||||
flow.id.hyphenated().to_string()
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
sqlx::query!(
|
||||
"UPDATE v2_job_queue SET running = false, started_at = null
|
||||
WHERE id = $1 AND canceled_by IS NULL",
|
||||
flow.id
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
} else {
|
||||
let id = flow.id.clone();
|
||||
let last_ping = flow.last_ping.clone();
|
||||
let now = now_from_db(db).await?;
|
||||
let reason = format!(
|
||||
"{} was hanging in between 2 steps. Last ping: {last_ping:?} (now: {now})",
|
||||
if flow.is_flow_step.unwrap_or(false) && flow.parent_job.is_some() {
|
||||
format!("Flow was cancelled because subflow {id}")
|
||||
} else {
|
||||
format!("Flow {id} was cancelled because it")
|
||||
}
|
||||
);
|
||||
report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await;
|
||||
cancel_zombie_flow_job(db, flow.id, &flow.workspace_id, reason).await?;
|
||||
}
|
||||
}
|
||||
|
||||
let flows2 = sqlx::query!(
|
||||
r#"
|
||||
DELETE
|
||||
FROM parallel_monitor_lock
|
||||
WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
|
||||
RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q
|
||||
WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL
|
||||
) AS workspace_id
|
||||
"#,
|
||||
FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str()
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
for flow in flows2 {
|
||||
if let Some(parent_flow_workspace_id) = flow.workspace_id {
|
||||
tracing::error!(
|
||||
"parallel Zombie flow detected: {} in workspace {}. Last ping was: {:?}.",
|
||||
flow.parent_flow_id,
|
||||
parent_flow_workspace_id,
|
||||
flow.last_ping
|
||||
);
|
||||
cancel_zombie_flow_job(db, flow.parent_flow_id, &parent_flow_workspace_id,
|
||||
format!("Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ", flow.parent_flow_id, flow.job_id))
|
||||
.await?;
|
||||
} else {
|
||||
tracing::info!("releasing lock for parallel flow: {}", flow.parent_flow_id);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cancel_zombie_flow_job(
|
||||
db: &Pool<Postgres>,
|
||||
id: Uuid,
|
||||
workspace_id: &str,
|
||||
message: String,
|
||||
) -> Result<(), error::Error> {
|
||||
let mut tx = db.begin().await?;
|
||||
tracing::error!(
|
||||
"zombie flow detected: {} in workspace {}. Cancelling it.",
|
||||
id,
|
||||
workspace_id
|
||||
);
|
||||
(tx, _) = cancel_job(
|
||||
"monitor",
|
||||
Some(message),
|
||||
id,
|
||||
workspace_id,
|
||||
tx,
|
||||
db,
|
||||
true,
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn reload_hub_base_url_setting(db: &DB, server_mode: bool) -> error::Result<()> {
|
||||
let hub_base_url = load_value_from_global_settings(db, HUB_BASE_URL_SETTING).await?;
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use serde::de::DeserializeOwned;
|
||||
use std::future::Future;
|
||||
use std::{str::FromStr, sync::Arc};
|
||||
use std::{default::Default, str::FromStr, sync::Arc};
|
||||
use windmill_api_client::types::{NewScript, ScriptLang as NewScriptLanguage};
|
||||
|
||||
#[cfg(feature = "enterprise")]
|
||||
@@ -13,7 +13,6 @@ use serde_json::json;
|
||||
use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres};
|
||||
|
||||
use tokio::sync::RwLock;
|
||||
#[cfg(feature = "enterprise")]
|
||||
use tokio::time::{timeout, Duration};
|
||||
|
||||
use windmill_api_client::types::{CreateFlowBody, RawScript};
|
||||
@@ -42,7 +41,7 @@ pub struct CompletedJob {
|
||||
pub parent_job: Option<Uuid>,
|
||||
pub created_by: String,
|
||||
pub created_at: chrono::DateTime<chrono::Utc>,
|
||||
pub started_at: chrono::DateTime<chrono::Utc>,
|
||||
pub started_at: Option<chrono::DateTime<chrono::Utc>>,
|
||||
pub duration_ms: i64,
|
||||
pub success: bool,
|
||||
pub script_path: Option<String>,
|
||||
@@ -153,7 +152,7 @@ impl ApiServer {
|
||||
println!("closing api server");
|
||||
let Self { tx, task, .. } = self;
|
||||
drop(tx);
|
||||
task.await.unwrap()
|
||||
task.await?
|
||||
}
|
||||
}
|
||||
|
||||
@@ -213,7 +212,7 @@ mod suspend_resume {
|
||||
}
|
||||
|
||||
fn flow() -> FlowValue {
|
||||
serde_json::from_value(serde_json::json!({
|
||||
serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"id": "a",
|
||||
"value": {
|
||||
@@ -302,7 +301,10 @@ mod suspend_resume {
|
||||
let queue = listen_for_queue(&db).await;
|
||||
let db_ = db.clone();
|
||||
|
||||
in_test_worker(&db, async move {
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
let db = db_;
|
||||
|
||||
wait_until_flow_suspends(flow, queue, &db).await;
|
||||
@@ -336,8 +338,9 @@ mod suspend_resume {
|
||||
.unwrap();
|
||||
|
||||
completed.find(&flow).await.unwrap();
|
||||
}, port)
|
||||
.await;
|
||||
}
|
||||
)
|
||||
.await;
|
||||
|
||||
server.close().await.unwrap();
|
||||
|
||||
@@ -409,7 +412,10 @@ mod suspend_resume {
|
||||
let queue = listen_for_queue(&db).await;
|
||||
let db_ = db.clone();
|
||||
|
||||
in_test_worker(&db, async move {
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
let db = db_;
|
||||
|
||||
wait_until_flow_suspends(flow, queue, &db).await;
|
||||
@@ -439,8 +445,9 @@ mod suspend_resume {
|
||||
.unwrap();
|
||||
|
||||
completed.find(&flow).await.unwrap();
|
||||
}, port)
|
||||
.await;
|
||||
}
|
||||
)
|
||||
.await;
|
||||
|
||||
server.close().await.unwrap();
|
||||
|
||||
@@ -496,7 +503,7 @@ mod retry {
|
||||
results
|
||||
});
|
||||
|
||||
return Self { addr, tx, task };
|
||||
Self { addr, tx, task }
|
||||
}
|
||||
|
||||
async fn close(self) -> Vec<u8> {
|
||||
@@ -528,7 +535,7 @@ def main(last, port):
|
||||
}
|
||||
|
||||
fn flow_forloop_retry() -> FlowValue {
|
||||
serde_json::from_value(serde_json::json!({
|
||||
serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"id": "a",
|
||||
"value": {
|
||||
@@ -640,15 +647,15 @@ def main(last, port):
|
||||
|
||||
assert_eq!(server.close().await, attempts);
|
||||
|
||||
assert!(
|
||||
assert_eq!(
|
||||
result[1]["error"]
|
||||
.as_object()
|
||||
.unwrap()
|
||||
.get("message")
|
||||
.unwrap()
|
||||
.as_str()
|
||||
.unwrap()
|
||||
== "read"
|
||||
.unwrap(),
|
||||
"read"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -775,7 +782,7 @@ async fn test_iteration(db: Pool<Postgres>) {
|
||||
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
@@ -806,7 +813,7 @@ async fn test_iteration(db: Pool<Postgres>) {
|
||||
.await
|
||||
.json_result()
|
||||
.unwrap();
|
||||
assert_eq!(result, serde_json::json!([]));
|
||||
assert_eq!(result, json!([]));
|
||||
|
||||
/* Don't actually test that this does 257 jobs or that will take forever. */
|
||||
let result =
|
||||
@@ -833,7 +840,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
|
||||
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
@@ -865,7 +872,7 @@ async fn test_iteration_parallel(db: Pool<Postgres>) {
|
||||
.await
|
||||
.json_result()
|
||||
.unwrap();
|
||||
assert_eq!(result, serde_json::json!([]));
|
||||
assert_eq!(result, json!([]));
|
||||
|
||||
/* Don't actually test that this does 257 jobs or that will take forever. */
|
||||
let job =
|
||||
@@ -946,7 +953,12 @@ impl RunJob {
|
||||
async fn run_until_complete(self, db: &Pool<Postgres>, port: u16) -> CompletedJob {
|
||||
let uuid = self.push(db).await;
|
||||
let listener = listen_for_completed_jobs(db).await;
|
||||
in_test_worker(db, listener.find(&uuid), port).await;
|
||||
in_test_worker(
|
||||
db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
listener.find(&uuid),
|
||||
)
|
||||
.await;
|
||||
let r = completed_job(uuid, db).await;
|
||||
r
|
||||
}
|
||||
@@ -961,7 +973,12 @@ impl RunJob {
|
||||
let uuid = self.push(db).await;
|
||||
let listener = listen_for_completed_jobs(db).await;
|
||||
test(uuid).await;
|
||||
in_test_worker(db, listener.find(&uuid), port).await;
|
||||
in_test_worker(
|
||||
db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
listener.find(&uuid),
|
||||
)
|
||||
.await;
|
||||
let r = completed_job(uuid, db).await;
|
||||
r
|
||||
}
|
||||
@@ -975,17 +992,34 @@ async fn run_job_in_new_worker_until_complete(
|
||||
RunJob::from(job).run_until_complete(db, port).await
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
enum MonitorZombiesOption {
|
||||
#[default]
|
||||
Disabled,
|
||||
Enabled {
|
||||
restart_zombies: bool,
|
||||
timeout_seconds: &'static str,
|
||||
flow_timeout_seconds: &'static str,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct WorkerOptions {
|
||||
port: u16,
|
||||
monitor_zombies: MonitorZombiesOption,
|
||||
}
|
||||
|
||||
/// Start a worker with a timeout and run a future, until the worker quits or we time out.
|
||||
///
|
||||
/// Cleans up the worker before resolving.
|
||||
async fn in_test_worker<Fut: std::future::Future>(
|
||||
async fn in_test_worker<Fut: Future>(
|
||||
db: &Pool<Postgres>,
|
||||
options: WorkerOptions,
|
||||
inner: Fut,
|
||||
port: u16,
|
||||
) -> <Fut as std::future::Future>::Output {
|
||||
) -> <Fut as Future>::Output {
|
||||
set_jwt_secret().await;
|
||||
let (quit, worker) = spawn_test_worker(db, port);
|
||||
let worker = tokio::time::timeout(std::time::Duration::from_secs(60), worker);
|
||||
let (quit, worker) = spawn_test_worker(db, options);
|
||||
let worker = timeout(Duration::from_secs(60), worker);
|
||||
tokio::pin!(worker);
|
||||
|
||||
let res = tokio::select! {
|
||||
@@ -1010,7 +1044,7 @@ async fn in_test_worker<Fut: std::future::Future>(
|
||||
|
||||
fn spawn_test_worker(
|
||||
db: &Pool<Postgres>,
|
||||
port: u16,
|
||||
options: WorkerOptions,
|
||||
) -> (
|
||||
tokio::sync::broadcast::Sender<()>,
|
||||
tokio::task::JoinHandle<()>,
|
||||
@@ -1032,8 +1066,9 @@ fn spawn_test_worker(
|
||||
let ip: &str = Default::default();
|
||||
|
||||
let tx2 = tx.clone();
|
||||
let future = async move {
|
||||
let base_internal_url = format!("http://localhost:{}", port);
|
||||
let db2 = db.clone();
|
||||
let worker_fut = async move {
|
||||
let base_internal_url = format!("http://localhost:{}", options.port);
|
||||
{
|
||||
let mut wc = WORKER_CONFIG.write().await;
|
||||
(*wc).worker_tags = windmill_common::worker::DEFAULT_TAGS.clone();
|
||||
@@ -1045,7 +1080,7 @@ fn spawn_test_worker(
|
||||
windmill_common::worker::make_pull_query(&wc).await;
|
||||
}
|
||||
windmill_worker::run_worker(
|
||||
&db,
|
||||
&db2,
|
||||
worker_instance,
|
||||
worker_name,
|
||||
1,
|
||||
@@ -1059,6 +1094,35 @@ fn spawn_test_worker(
|
||||
.await
|
||||
};
|
||||
|
||||
let MonitorZombiesOption::Enabled { restart_zombies, timeout_seconds, flow_timeout_seconds } =
|
||||
options.monitor_zombies
|
||||
else {
|
||||
return (tx, tokio::task::spawn(worker_fut));
|
||||
};
|
||||
|
||||
let mut rx = tx.subscribe();
|
||||
let monitor_fut = async move {
|
||||
let base_internal_url = format!("http://localhost:{}", options.port);
|
||||
loop {
|
||||
if rx.try_recv().is_ok() {
|
||||
break;
|
||||
}
|
||||
windmill_worker::monitor::monitor_once(
|
||||
&db,
|
||||
&base_internal_url,
|
||||
worker_instance,
|
||||
Some(timeout_seconds),
|
||||
Some(restart_zombies),
|
||||
Some(flow_timeout_seconds),
|
||||
)
|
||||
.await;
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
};
|
||||
|
||||
let future = async move {
|
||||
tokio::select! { biased; _ = worker_fut => (), _ = monitor_fut => () }
|
||||
};
|
||||
(tx, tokio::task::spawn(future))
|
||||
}
|
||||
|
||||
@@ -1101,10 +1165,10 @@ async fn completed_job(uuid: Uuid, db: &Pool<Postgres>) -> CompletedJob {
|
||||
}
|
||||
|
||||
#[axum::async_trait(?Send)]
|
||||
trait StreamFind: futures::Stream + Unpin + Sized {
|
||||
trait StreamFind: Stream + Unpin + Sized {
|
||||
async fn find(self, item: &Self::Item) -> Option<Self::Item>
|
||||
where
|
||||
for<'l> &'l Self::Item: std::cmp::PartialEq,
|
||||
for<'l> &'l Self::Item: PartialEq,
|
||||
{
|
||||
use futures::{future::ready, StreamExt};
|
||||
|
||||
@@ -1112,7 +1176,7 @@ trait StreamFind: futures::Stream + Unpin + Sized {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: futures::Stream + Unpin + Sized> StreamFind for T {}
|
||||
impl<T: Stream + Unpin + Sized> StreamFind for T {}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn test_deno_flow(db: Pool<Postgres>) {
|
||||
@@ -1228,7 +1292,7 @@ async fn test_deno_flow(db: Pool<Postgres>) {
|
||||
let job = run_job_in_new_worker_until_complete(&db, job.clone(), port).await;
|
||||
// println!("job: {:#?}", job.flow_status);
|
||||
let result = job.json_result().unwrap();
|
||||
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {}", i);
|
||||
assert_eq!(result, json!([2, 4, 6]), "iteration: {}", i);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1238,7 +1302,7 @@ async fn test_identity(db: Pool<Postgres>) {
|
||||
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"value": {
|
||||
"type": "rawscript",
|
||||
@@ -1266,7 +1330,7 @@ async fn test_identity(db: Pool<Postgres>) {
|
||||
.await
|
||||
.json_result()
|
||||
.unwrap();
|
||||
assert_eq!(result, serde_json::json!(42));
|
||||
assert_eq!(result, json!(42));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -1488,7 +1552,7 @@ async fn test_deno_flow_same_worker(db: Pool<Postgres>) {
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
result,
|
||||
serde_json::json!("false 1,true 1,false 1,true 2,false 1,true 3,false 1,true 3")
|
||||
json!("false 1,true 1,false 1,true 2,false 1,true 3,false 1,true 3")
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1541,7 +1605,7 @@ async fn test_flow_result_by_id(db: Pool<Postgres>) {
|
||||
.await
|
||||
.json_result()
|
||||
.unwrap();
|
||||
assert_eq!(result, serde_json::json!([[42]]));
|
||||
assert_eq!(result, json!([[42]]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -1551,7 +1615,7 @@ async fn test_stop_after_if(db: Pool<Postgres>) {
|
||||
// let port = server.addr.port();
|
||||
|
||||
let port = 123;
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
@@ -1604,7 +1668,7 @@ async fn test_stop_after_if_nested(db: Pool<Postgres>) {
|
||||
// let port = server.addr.port();
|
||||
|
||||
let port = 123;
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
@@ -1664,7 +1728,7 @@ async fn test_python_flow(db: Pool<Postgres>) {
|
||||
let numbers = "def main(): return [1, 2, 3]";
|
||||
let doubles = "def main(n): return n * 2";
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!( {
|
||||
let flow: FlowValue = serde_json::from_value(json!( {
|
||||
"modules": [
|
||||
{
|
||||
"value": {
|
||||
@@ -1708,7 +1772,7 @@ async fn test_python_flow(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([2, 4, 6]), "iteration: {i}");
|
||||
assert_eq!(result, json!([2, 4, 6]), "iteration: {i}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1718,7 +1782,7 @@ async fn test_python_flow_2(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"value": {
|
||||
@@ -1743,7 +1807,7 @@ async fn test_python_flow_2(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("Hello"), "iteration: {i}");
|
||||
assert_eq!(result, json!("Hello"), "iteration: {i}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1783,7 +1847,7 @@ func main(derp string) (string, error) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("hello world"));
|
||||
assert_eq!(result, json!("hello world"));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -1818,7 +1882,7 @@ fn main(world: String) -> Result<String, String> {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("Hello Hyrule!"));
|
||||
assert_eq!(result, json!("Hello Hyrule!"));
|
||||
}
|
||||
|
||||
// #[sqlx::test(fixtures("base"))]
|
||||
@@ -1923,7 +1987,7 @@ def main():
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("hello world"));
|
||||
assert_eq!(result, json!("hello world"));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -1959,7 +2023,7 @@ def main():
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!(3));
|
||||
assert_eq!(result, json!(3));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -1994,7 +2058,7 @@ def main():
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("test-workspace"));
|
||||
assert_eq!(result, json!("test-workspace"));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2028,7 +2092,7 @@ export async function main(a: Date) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("object"));
|
||||
assert_eq!(result, json!("object"));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2062,7 +2126,7 @@ export async function main(a: Date) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!("object"));
|
||||
assert_eq!(result, json!("object"));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2097,7 +2161,7 @@ def main(a: datetime, b: bytes):
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([true, true]));
|
||||
assert_eq!(result, json!([true, true]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2106,7 +2170,7 @@ async fn test_empty_loop_1(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
@@ -2153,7 +2217,7 @@ async fn test_empty_loop_1(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!(0));
|
||||
assert_eq!(result, json!(0));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2162,7 +2226,7 @@ async fn test_invalid_first_step(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"value": {
|
||||
@@ -2200,7 +2264,7 @@ async fn test_empty_loop_2(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"value": {
|
||||
@@ -2233,7 +2297,7 @@ async fn test_empty_loop_2(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([]));
|
||||
assert_eq!(result, json!([]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2241,7 +2305,7 @@ async fn test_step_after_loop(db: Pool<Postgres>) {
|
||||
initialize_tracing().await;
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [
|
||||
{
|
||||
"id": "a",
|
||||
@@ -2288,7 +2352,7 @@ async fn test_step_after_loop(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!(9));
|
||||
assert_eq!(result, json!(9));
|
||||
}
|
||||
|
||||
fn module_add_item_to_list(i: i32, id: &str) -> serde_json::Value {
|
||||
@@ -2356,7 +2420,7 @@ async fn test_branchone_simple(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([1, 2]));
|
||||
assert_eq!(result, json!([1, 2]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2392,7 +2456,7 @@ async fn test_branchone_with_cond(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([1, 3]));
|
||||
assert_eq!(result, json!([1, 3]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2430,7 +2494,7 @@ async fn test_branchall_sequential(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
|
||||
assert_eq!(result, json!([[1, 2], [1, 3]]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2467,7 +2531,7 @@ async fn test_branchall_simple(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([[1, 2], [1, 3]]));
|
||||
assert_eq!(result, json!([[1, 2], [1, 3]]));
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -2615,7 +2679,7 @@ async fn test_branchone_nested(db: Pool<Postgres>) {
|
||||
.json_result()
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, serde_json::json!([1, 2, 3]));
|
||||
assert_eq!(result, json!([1, 2, 3]));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
@@ -2675,7 +2739,7 @@ async fn test_branchall_nested(db: Pool<Postgres>) {
|
||||
println!("{:#?}", result);
|
||||
assert_eq!(
|
||||
result,
|
||||
serde_json::json!([[[[1, 2], [1, 3], 4], [[1, 2], [1, 3], 5]], [1, 6]])
|
||||
json!([[[[1, 2], [1, 3], 4], [[1, 2], [1, 3], 5]], [1, 6]])
|
||||
);
|
||||
}
|
||||
|
||||
@@ -2685,7 +2749,7 @@ async fn test_failure_module(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: FlowValue = serde_json::from_value(serde_json::json!({
|
||||
let flow: FlowValue = serde_json::from_value(json!({
|
||||
"modules": [{
|
||||
"id": "a",
|
||||
"value": {
|
||||
@@ -2798,7 +2862,7 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
|
||||
let server = ApiServer::start(db.clone()).await;
|
||||
let port = server.addr.port();
|
||||
|
||||
let flow: windmill_api_client::types::OpenFlow = serde_json::from_value(serde_json::json!({
|
||||
let flow: windmill_api_client::types::OpenFlow = serde_json::from_value(json!({
|
||||
"summary": "",
|
||||
"description": "",
|
||||
"value": {
|
||||
@@ -2898,7 +2962,12 @@ async fn test_flow_lock_all(db: Pool<Postgres>) {
|
||||
.unwrap();
|
||||
let mut str = listen_for_completed_jobs(&db).await;
|
||||
let listen_first_job = str.next();
|
||||
in_test_worker(&db, listen_first_job, port).await;
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
listen_first_job,
|
||||
)
|
||||
.await;
|
||||
|
||||
let modules = client
|
||||
.get_flow_by_path("test-workspace", "g/all/flow_lock_all", None)
|
||||
@@ -3193,6 +3262,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
str.next().await; // completed error job
|
||||
|
||||
@@ -3220,7 +3290,6 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
|
||||
);
|
||||
}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -3264,6 +3333,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
str.next().await; // completed working job
|
||||
let uuid = timeout(Duration::from_millis(5000), str.next()).await; // recovery handler
|
||||
@@ -3287,7 +3357,6 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
|
||||
panic!("a script was run after main job execution but was not schedule recovery handler");
|
||||
}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
@@ -3346,6 +3415,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
str.next().await; // completed error step
|
||||
str.next().await; // completed error flow
|
||||
@@ -3374,7 +3444,6 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
|
||||
);
|
||||
}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -3418,6 +3487,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
str.next().await; // completed working step
|
||||
str.next().await; // completed working flow
|
||||
@@ -3442,7 +3512,6 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
|
||||
panic!("a script was run after main job execution but was not schedule recovery handler");
|
||||
}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
@@ -3502,6 +3571,7 @@ async fn run_deployed_relative_imports(
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
completed.next().await; // deployed script
|
||||
|
||||
@@ -3534,7 +3604,7 @@ async fn run_deployed_relative_imports(
|
||||
|
||||
assert_eq!(
|
||||
result,
|
||||
serde_json::json!([
|
||||
json!([
|
||||
"f/system/same_folder_script",
|
||||
"f/system/same_folder_script",
|
||||
"f/system_relative/different_folder_script",
|
||||
@@ -3542,7 +3612,6 @@ async fn run_deployed_relative_imports(
|
||||
])
|
||||
);
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
@@ -3560,6 +3629,7 @@ async fn run_preview_relative_imports(
|
||||
let db2 = db.clone();
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
let job = RunJob::from(JobPayload::Code(RawCode {
|
||||
hash: None,
|
||||
@@ -3582,7 +3652,7 @@ async fn run_preview_relative_imports(
|
||||
|
||||
assert_eq!(
|
||||
result,
|
||||
serde_json::json!([
|
||||
json!([
|
||||
"f/system/same_folder_script",
|
||||
"f/system/same_folder_script",
|
||||
"f/system_relative/different_folder_script",
|
||||
@@ -3590,7 +3660,6 @@ async fn run_preview_relative_imports(
|
||||
])
|
||||
);
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
@@ -3864,6 +3933,7 @@ async fn test_workflow_as_code(db: Pool<Postgres>) {
|
||||
let db = &db;
|
||||
in_test_worker(
|
||||
&db,
|
||||
WorkerOptions { port, ..Default::default() },
|
||||
async move {
|
||||
let job = RunJob::from(JobPayload::Code(RawCode {
|
||||
language: ScriptLang::Python3,
|
||||
@@ -3912,7 +3982,6 @@ async fn test_workflow_as_code(db: Pool<Postgres>) {
|
||||
);
|
||||
}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ mod handle_child;
|
||||
mod job_logger;
|
||||
mod job_logger_ee;
|
||||
mod js_eval;
|
||||
pub mod monitor;
|
||||
#[cfg(feature = "mysql")]
|
||||
mod mysql_executor;
|
||||
#[cfg(feature = "oracledb")]
|
||||
|
||||
450
backend/windmill-worker/src/monitor.rs
Normal file
450
backend/windmill-worker/src/monitor.rs
Normal file
@@ -0,0 +1,450 @@
|
||||
use std::{
|
||||
env,
|
||||
sync::{atomic::AtomicU16, Arc},
|
||||
};
|
||||
|
||||
use sqlx::{PgPool, Pool, Postgres};
|
||||
use tokio::sync::mpsc;
|
||||
use uuid::Uuid;
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
use windmill_common::METRICS_ENABLED;
|
||||
use windmill_common::{
|
||||
error,
|
||||
flow_status::{FlowStatus, FlowStatusModule},
|
||||
jobs::QueuedJob,
|
||||
utils::{now_from_db, report_critical_error},
|
||||
BASE_URL,
|
||||
};
|
||||
use windmill_queue::cancel_job;
|
||||
|
||||
use crate::{
|
||||
create_token_for_owner, handle_job_error, AuthedClient, SameWorkerPayload, SameWorkerSender,
|
||||
SendResult, SCRIPT_TOKEN_EXPIRY,
|
||||
};
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref ZOMBIE_JOB_TIMEOUT: String = env::var("ZOMBIE_JOB_TIMEOUT")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<String>().ok())
|
||||
.unwrap_or_else(|| "60".to_string());
|
||||
|
||||
static ref FLOW_ZOMBIE_TRANSITION_TIMEOUT: String = env::var("FLOW_ZOMBIE_TRANSITION_TIMEOUT")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<String>().ok())
|
||||
.unwrap_or_else(|| "60".to_string());
|
||||
|
||||
pub static ref RESTART_ZOMBIE_JOBS: bool = env::var("RESTART_ZOMBIE_JOBS")
|
||||
.ok()
|
||||
.and_then(|x| x.parse::<bool>().ok())
|
||||
.unwrap_or(true);
|
||||
}
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
lazy_static::lazy_static! {
|
||||
static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter =
|
||||
prometheus::register_int_counter!(
|
||||
"queue_zombie_restart_count",
|
||||
"Total number of jobs restarted due to ping timeout."
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter =
|
||||
prometheus::register_int_counter!(
|
||||
"queue_zombie_delete_count",
|
||||
"Total number of jobs deleted due to their ping timing out in an unrecoverable state."
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
static ref QUEUE_COUNT: prometheus::IntGaugeVec =
|
||||
prometheus::register_int_gauge_vec!(
|
||||
"queue_count",
|
||||
"Number of jobs in the queue",
|
||||
&["tag"]
|
||||
).unwrap();
|
||||
}
|
||||
|
||||
pub async fn monitor_once(
|
||||
db: &PgPool,
|
||||
base_internal_url: &str,
|
||||
worker_name: &str,
|
||||
timeout_seconds: Option<&str>,
|
||||
restart_zombies: Option<bool>,
|
||||
transition_timeout_seconds: Option<&str>,
|
||||
) {
|
||||
let timeout_seconds = timeout_seconds.unwrap_or_else(|| &*ZOMBIE_JOB_TIMEOUT);
|
||||
let restart_zombies = restart_zombies.unwrap_or(*RESTART_ZOMBIE_JOBS);
|
||||
handle_zombie_jobs(
|
||||
&db,
|
||||
base_internal_url,
|
||||
worker_name,
|
||||
timeout_seconds,
|
||||
restart_zombies,
|
||||
)
|
||||
.await;
|
||||
let transition_timeout_seconds =
|
||||
transition_timeout_seconds.unwrap_or_else(|| &*FLOW_ZOMBIE_TRANSITION_TIMEOUT);
|
||||
let _ = handle_zombie_flows(&db, transition_timeout_seconds)
|
||||
.await
|
||||
.inspect_err(|err| tracing::error!("Error handling zombie flows: {:#}", err));
|
||||
}
|
||||
|
||||
async fn handle_zombie_jobs(
|
||||
db: &Pool<Postgres>,
|
||||
base_internal_url: &str,
|
||||
worker_name: &str,
|
||||
timeout_seconds: &str,
|
||||
restart_zombies: bool,
|
||||
) {
|
||||
if restart_zombies {
|
||||
let restarted = sqlx::query!(
|
||||
"WITH zombie_jobs AS (
|
||||
UPDATE v2_job_queue q SET running = false, started_at = null
|
||||
FROM v2_job j, v2_job_runtime r
|
||||
WHERE j.id = q.id AND j.id = r.id
|
||||
AND ping < now() - ($1 || ' seconds')::interval
|
||||
AND running = true
|
||||
AND kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')
|
||||
AND same_worker = false
|
||||
RETURNING q.id, q.workspace_id, ping
|
||||
),
|
||||
update_concurrency AS (
|
||||
UPDATE concurrency_counter cc
|
||||
SET job_uuids = job_uuids - zj.id::text
|
||||
FROM zombie_jobs zj
|
||||
INNER JOIN concurrency_key ck ON ck.job_id = zj.id
|
||||
WHERE cc.concurrency_id = ck.key
|
||||
)
|
||||
SELECT id, workspace_id, ping FROM zombie_jobs",
|
||||
timeout_seconds,
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![]);
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
|
||||
}
|
||||
|
||||
let base_url = BASE_URL.read().await.clone();
|
||||
for r in restarted {
|
||||
let last_ping = r
|
||||
.ping
|
||||
.map(|x| format!("last ping at {x}"))
|
||||
.unwrap_or_else(|| "no last ping".into());
|
||||
let url = format!("{}/run/{}?workspace={}", base_url, r.id, r.workspace_id,);
|
||||
let error_message = format!(
|
||||
"Zombie job {} on {} ({}) detected, restarting it, {}",
|
||||
r.id, r.workspace_id, url, last_ping
|
||||
);
|
||||
|
||||
let _ = sqlx::query!("
|
||||
INSERT INTO job_logs (job_id, logs)
|
||||
VALUES ($1, 'Restarted job after not receiving job''s ping for too long the ' || now() || '\n\n')
|
||||
ON CONFLICT (job_id) DO UPDATE SET logs = job_logs.logs || '\n' || EXCLUDED.logs
|
||||
WHERE job_logs.job_id = $1",
|
||||
r.id
|
||||
)
|
||||
.execute(db)
|
||||
.await;
|
||||
tracing::error!(error_message);
|
||||
report_critical_error(error_message, db.clone(), Some(&r.workspace_id), None).await;
|
||||
}
|
||||
}
|
||||
|
||||
let same_worker_timeout_jobs = {
|
||||
let long_same_worker_jobs = sqlx::query!(
|
||||
"SELECT worker, array_agg(v2_job_queue.id) as ids
|
||||
FROM v2_job_queue
|
||||
JOIN v2_job USING (id)
|
||||
LEFT JOIN v2_job_runtime USING (id)
|
||||
WHERE v2_job_queue.created_at < now() - ('60 seconds')::interval
|
||||
AND running = true AND ping IS NULL
|
||||
AND same_worker = true
|
||||
AND worker IS NOT NULL
|
||||
GROUP BY worker",
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![]);
|
||||
|
||||
let worker_ids = long_same_worker_jobs
|
||||
.iter()
|
||||
.map(|x| x.worker.clone().unwrap_or_default())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let long_dead_workers: std::collections::HashSet<String> = sqlx::query_scalar!(
|
||||
"WITH worker_ids AS (SELECT unnest($1::text[]) as worker)
|
||||
SELECT worker_ids.worker FROM worker_ids
|
||||
LEFT JOIN worker_ping ON worker_ids.worker = worker_ping.worker
|
||||
WHERE worker_ping.worker IS NULL OR worker_ping.ping_at < now() - ('60 seconds')::interval",
|
||||
&worker_ids[..]
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![])
|
||||
.into_iter()
|
||||
.filter_map(|x| x)
|
||||
.collect();
|
||||
|
||||
let mut timeouts: Vec<Uuid> = vec![];
|
||||
for worker in long_same_worker_jobs {
|
||||
if worker.worker.is_some() && long_dead_workers.contains(&worker.worker.unwrap()) {
|
||||
if let Some(ids) = worker.ids {
|
||||
timeouts.extend(ids);
|
||||
}
|
||||
}
|
||||
}
|
||||
if !timeouts.is_empty() {
|
||||
tracing::error!(
|
||||
"Failing same worker zombie jobs: {:?}",
|
||||
timeouts
|
||||
.iter()
|
||||
.map(|x| x.hyphenated().to_string())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
);
|
||||
}
|
||||
|
||||
let jobs = sqlx::query_as::<_, QueuedJob>("SELECT * FROM v2_as_queue WHERE id = ANY($1)")
|
||||
.bind(&timeouts[..])
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.map_err(|e| tracing::error!("Error fetching same worker jobs: {:?}", e))
|
||||
.unwrap_or_default();
|
||||
|
||||
jobs
|
||||
};
|
||||
|
||||
let non_restartable_jobs = if restart_zombies {
|
||||
vec![]
|
||||
} else {
|
||||
sqlx::query_as::<_, QueuedJob>(
|
||||
"SELECT * FROM v2_as_queue
|
||||
WHERE last_ping < now() - ($1 || ' seconds')::interval
|
||||
AND running = true
|
||||
AND job_kind NOT IN ('flow', 'flowpreview', 'flownode', 'singlescriptflow')
|
||||
AND same_worker = false",
|
||||
)
|
||||
.bind(timeout_seconds)
|
||||
.fetch_all(db)
|
||||
.await
|
||||
.ok()
|
||||
.unwrap_or_else(|| vec![])
|
||||
};
|
||||
|
||||
let timeouts = non_restartable_jobs
|
||||
.into_iter()
|
||||
.chain(same_worker_timeout_jobs)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
#[cfg(feature = "prometheus")]
|
||||
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
|
||||
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
|
||||
}
|
||||
|
||||
for job in timeouts {
|
||||
// since the job is unrecoverable, the same worker queue should never be sent anything
|
||||
let (same_worker_tx_never_used, _same_worker_rx_never_used) =
|
||||
mpsc::channel::<SameWorkerPayload>(1);
|
||||
let same_worker_tx_never_used =
|
||||
SameWorkerSender(same_worker_tx_never_used, Arc::new(AtomicU16::new(0)));
|
||||
let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::<SendResult>(1);
|
||||
|
||||
let label = if job.permissioned_as != format!("u/{}", job.created_by)
|
||||
&& job.permissioned_as != job.created_by
|
||||
{
|
||||
format!("ephemeral-script-end-user-{}", job.created_by)
|
||||
} else {
|
||||
"ephemeral-script".to_string()
|
||||
};
|
||||
let token = create_token_for_owner(
|
||||
&db,
|
||||
&job.workspace_id,
|
||||
&job.permissioned_as,
|
||||
&label,
|
||||
*SCRIPT_TOKEN_EXPIRY,
|
||||
&job.email,
|
||||
&job.id,
|
||||
)
|
||||
.await
|
||||
.expect("could not create job token");
|
||||
|
||||
let client = AuthedClient {
|
||||
base_internal_url: base_internal_url.to_string(),
|
||||
token,
|
||||
workspace: job.workspace_id.to_string(),
|
||||
force_client: None,
|
||||
};
|
||||
|
||||
let last_ping = job.last_ping.clone();
|
||||
let _ = handle_job_error(
|
||||
db,
|
||||
&client,
|
||||
&job,
|
||||
0,
|
||||
None,
|
||||
error::Error::ExecutionErr(format!(
|
||||
"Job timed out after no ping from job since {} (timeout: {}, same_worker: {})",
|
||||
last_ping
|
||||
.map(|x| x.to_string())
|
||||
.unwrap_or_else(|| "no ping".to_string()),
|
||||
timeout_seconds,
|
||||
job.same_worker
|
||||
)),
|
||||
true,
|
||||
same_worker_tx_never_used,
|
||||
"",
|
||||
worker_name,
|
||||
send_result_never_used,
|
||||
#[cfg(feature = "benchmark")]
|
||||
&mut crate::bench::BenchmarkIter::new(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_zombie_flows(db: &PgPool, transition_timeout_seconds: &str) -> error::Result<()> {
|
||||
let flows = sqlx::query!(
|
||||
r#"
|
||||
SELECT
|
||||
id AS "id!", workspace_id AS "workspace_id!", parent_job, is_flow_step,
|
||||
flow_status AS "flow_status: Box<str>", last_ping, same_worker
|
||||
FROM v2_as_queue
|
||||
WHERE running = true AND suspend = 0 AND suspend_until IS null AND scheduled_for <= now()
|
||||
AND (job_kind = 'flow' OR job_kind = 'flowpreview' OR job_kind = 'flownode')
|
||||
AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
|
||||
AND canceled = false
|
||||
"#,
|
||||
transition_timeout_seconds
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
for flow in flows {
|
||||
let status = flow
|
||||
.flow_status
|
||||
.as_deref()
|
||||
.and_then(|x| serde_json::from_str::<FlowStatus>(x).ok());
|
||||
if !flow.same_worker.unwrap_or(false)
|
||||
&& status.is_some_and(|s| {
|
||||
s.modules
|
||||
.get(0)
|
||||
.is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))
|
||||
})
|
||||
{
|
||||
let error_message = format!(
|
||||
"Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
|
||||
flow.id, flow.workspace_id
|
||||
);
|
||||
tracing::error!(error_message);
|
||||
report_critical_error(error_message, db.clone(), Some(&flow.workspace_id), None).await;
|
||||
// if the flow hasn't started and is a zombie, we can simply restart it
|
||||
let mut tx = db.begin().await?;
|
||||
|
||||
let concurrency_key =
|
||||
sqlx::query_scalar!("SELECT key FROM concurrency_key WHERE job_id = $1", flow.id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
|
||||
if let Some(key) = concurrency_key {
|
||||
sqlx::query!(
|
||||
"UPDATE concurrency_counter SET job_uuids = job_uuids - $2 WHERE concurrency_id = $1",
|
||||
key,
|
||||
flow.id.hyphenated().to_string()
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
sqlx::query!(
|
||||
"UPDATE v2_job_queue SET running = false, started_at = null
|
||||
WHERE id = $1 AND canceled_by IS NULL",
|
||||
flow.id
|
||||
)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
|
||||
tx.commit().await?;
|
||||
} else {
|
||||
let id = flow.id.clone();
|
||||
let last_ping = flow.last_ping.clone();
|
||||
let now = now_from_db(db).await?;
|
||||
let reason = format!(
|
||||
"{} was hanging in between 2 steps. Last ping: {last_ping:?} (now: {now})",
|
||||
if flow.is_flow_step.unwrap_or(false) && flow.parent_job.is_some() {
|
||||
format!("Flow was cancelled because subflow {id}")
|
||||
} else {
|
||||
format!("Flow {id} was cancelled because it")
|
||||
}
|
||||
);
|
||||
report_critical_error(reason.clone(), db.clone(), Some(&flow.workspace_id), None).await;
|
||||
cancel_zombie_flow_job(db, flow.id, &flow.workspace_id, reason).await?;
|
||||
}
|
||||
}
|
||||
|
||||
let flows2 = sqlx::query!(
|
||||
r#"
|
||||
DELETE
|
||||
FROM parallel_monitor_lock
|
||||
WHERE last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
|
||||
RETURNING parent_flow_id, job_id, last_ping, (SELECT workspace_id FROM v2_job_queue q
|
||||
WHERE q.id = parent_flow_id AND q.running = true AND q.canceled_by IS NULL
|
||||
) AS workspace_id
|
||||
"#,
|
||||
transition_timeout_seconds
|
||||
)
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
|
||||
for flow in flows2 {
|
||||
if let Some(parent_flow_workspace_id) = flow.workspace_id {
|
||||
tracing::error!(
|
||||
"parallel Zombie flow detected: {} in workspace {}. Last ping was: {:?}.",
|
||||
flow.parent_flow_id,
|
||||
parent_flow_workspace_id,
|
||||
flow.last_ping
|
||||
);
|
||||
cancel_zombie_flow_job(db, flow.parent_flow_id, &parent_flow_workspace_id, format!(
|
||||
"Flow {} cancelled as one of the parallel branch {} was unable to make the last transition ",
|
||||
flow.parent_flow_id,
|
||||
flow.job_id
|
||||
))
|
||||
.await?;
|
||||
} else {
|
||||
tracing::info!("releasing lock for parallel flow: {}", flow.parent_flow_id);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn cancel_zombie_flow_job(
|
||||
db: &Pool<Postgres>,
|
||||
id: Uuid,
|
||||
workspace_id: &str,
|
||||
message: String,
|
||||
) -> Result<(), error::Error> {
|
||||
let mut tx = db.begin().await?;
|
||||
tracing::error!(
|
||||
"zombie flow detected: {} in workspace {}. Cancelling it.",
|
||||
id,
|
||||
workspace_id
|
||||
);
|
||||
(tx, _) = cancel_job(
|
||||
"monitor",
|
||||
Some(message),
|
||||
id,
|
||||
workspace_id,
|
||||
tx,
|
||||
db,
|
||||
true,
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user