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

feat: more resilient flows in case of crash during transitions

Browse Source
This commit is contained in:
Ruben Fiszel
2024-02-25 18:25:40 +01:00
parent 3b797db8f4
commit efbba24581
16 changed files with 262 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.cargo/config.toml
-4
View File
@@ -1,8 +1,4 @@
[build]
rustflags = [
"--cfg",
"tokio_unstable"
]
incremental = true
[target.x86_64-apple-darwin]
backend/.sqlx/query-019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET last_ping = null\n WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "019edf515dbc2c04960b29b84ee4d6c44132758a6ff14f76485b7b089db25d47"
}
backend/.sqlx/query-45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET last_ping = now() WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": []
},
"hash": "45c9ecf8b1f8cbca7c75dab24a1eb6da8ceb45258ee5817ec71e73bebbe415bd"
}
backend/.sqlx/query-631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue\n SET last_ping = null\n WHERE id = $1 AND last_ping = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Timestamptz"
]
},
"nullable": []
},
"hash": "631d4637e4137a0680ffa56e4639c009214eeaee8fd27a0d3050998b354c45ff"
}
backend/.sqlx/query-003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91.json → backend/.sqlx/query-7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping",
"query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' \n WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping",
"describe": {
"columns": [
{
@@ -30,5 +30,5 @@
false
]
},
"hash": "003565f92aebec443c91f5c1c2a5953b98ee7f91d7c3b34947ab8ef1f2252a91"
"hash": "7ab02d2c060ed12531e6af98a5d3369afe534519bf48f61f88f60786db870c24"
}
backend/.sqlx/query-c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE queue SET running = false, started_at = null WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "c05be905e46c5b0a2186ba859a725495e55df7f2ad839aea22d0286525eb823e"
}
backend/Cargo.toml
+2
View File
@@ -46,11 +46,13 @@ pg_embed = ["dep:pg-embed"]
embedding = ["windmill-api/embedding"]
parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/parquet"]
prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill-worker/prometheus", "windmill-queue/prometheus"]
flow_testing = ["windmill-worker/flow_testing"]
[dependencies]
anyhow.workspace = true
tokio.workspace = true
dotenv.workspace = true
windmill-queue.workspace = true
windmill-common = { workspace = true, default-features = false }
windmill-git-sync.workspace = true
windmill-api = { workspace = true, default-features = false }
backend/migrations/20240225142058_make_last_ping_nullable.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20240225142058_make_last_ping_nullable.up.sql
+3
View File
@@ -0,0 +1,3 @@
-- Add up migration script here
ALTER TABLE queue
ALTER COLUMN last_ping DROP NOT NULL;
backend/src/main.rs
+17 -5
View File
@@ -624,6 +624,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
})
.unwrap_or_else(|| rd_string(5));
#[cfg(tokio_unstable)]
let monitor = tokio_metrics::TaskMonitor::new();
let ip = windmill_common::external_ip::get_ip()
@@ -678,9 +679,11 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
let base_internal_url = base_internal_url.clone();
let rsmq2 = rsmq.clone();
let sync_barrier = sync_barrier.clone();
handles.push(tokio::spawn(monitor.instrument(async move {
handles.push(tokio::spawn(async move {
tracing::info!(worker = %worker_name, "starting worker");
windmill_worker::run_worker(
let f = windmill_worker::run_worker(
&db1,
&instance_name,
worker_name,
@@ -693,9 +696,18 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
rsmq2,
sync_barrier,
agent_mode,
)
.await
})));
);
#[cfg(tokio_unstable)]
{
monitor.monitor(f, "worker").await
}
#[cfg(not(tokio_unstable))]
{
f.await
}
}));
}
futures::future::try_join_all(handles).await?;
backend/src/monitor.rs
+91 -16
View File
@@ -7,6 +7,7 @@ use std::{
time::Duration,
};
use rsmq_async::MultiplexedRsmq;
use serde::de::DeserializeOwned;
use sqlx::{Pool, Postgres};
use tokio::{
@@ -19,25 +20,19 @@ use windmill_api::{
DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN,
};
use windmill_common::{
error,
global_settings::{
error, flow_status::FlowStatusModule, global_settings::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
jobs::QueuedJob,
oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
server::load_server_config,
users::truncate_token,
worker::{
}, jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_server_config, users::truncate_token, worker::{
load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG,
WORKER_CONFIG,
},
BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
}, BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED
};
use windmill_queue::cancel_job;
use windmill_worker::{
create_token_for_owner, handle_job_error, AuthedClient, SendResult, BUNFIG_INSTALL_SCOPES,
JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, NPM_CONFIG_REGISTRY, PIP_EXTRA_INDEX_URL, PIP_INDEX_URL,
@@ -58,6 +53,11 @@ lazy_static::lazy_static! {
.and_then(|x| x.parse::<String>().ok())
.unwrap_or_else(|| "30".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(|| "10".to_string());
pub static ref RESTART_ZOMBIE_JOBS: bool = std::env::var("RESTART_ZOMBIE_JOBS")
.ok()
@@ -566,15 +566,21 @@ pub async fn monitor_pool(db: &DB) {
}
}
pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
pub async fn monitor_db(
db: &Pool<Postgres>,
base_internal_url: &str,
rsmq: Option<R>,
rsmq: Option<MultiplexedRsmq>,
server_mode: bool,
) {
let zombie_jobs_f = async {
if server_mode {
handle_zombie_jobs(db, base_internal_url, rsmq.clone(), "server").await;
match handle_zombie_flows(db, rsmq.clone()).await {
Err(err) => {
tracing::error!("Error handling zombie flows: {:?}", err);
}
_ => {}
}
}
};
let expired_items_f = async {
@@ -766,7 +772,8 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
) {
if *RESTART_ZOMBIE_JOBS {
let restarted = sqlx::query!(
"UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval
"UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n'
WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow') AND same_worker = false RETURNING id, workspace_id, last_ping",
*ZOMBIE_JOB_TIMEOUT,
)
@@ -779,8 +786,8 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
}
for r in restarted {
tracing::info!(
"restarted zombie job {} {} {}",
tracing::error!(
"Zombie job detected, restarting it: {} {} {:?}",
r.id,
r.workspace_id,
r.last_ping
@@ -788,7 +795,8 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
}
}
let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow')".to_string();
let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval
AND running = true AND job_kind NOT IN ('flow', 'flowpreview', 'singlescriptflow')".to_string();
if *RESTART_ZOMBIE_JOBS {
timeout_query.push_str(" AND same_worker = true");
};
@@ -853,3 +861,70 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await;
}
}
async fn handle_zombie_flows(
db: &DB,
rsmq: Option<rsmq_async::MultiplexedRsmq>,
) -> error::Result<()> {
let flows = sqlx::query_as::<_, QueuedJob>(
r#"
SELECT *
FROM queue
WHERE running = true AND suspend = 0 AND scheduled_for <= now() AND (job_kind = 'flow' OR job_kind = 'flowpreview')
AND last_ping IS NOT NULL AND last_ping < NOW() - ($1 || ' seconds')::interval
"#,
).bind(FLOW_ZOMBIE_TRANSITION_TIMEOUT.as_str())
.fetch_all(db)
.await?;
// TODO: for now only log zombie flows
let mut tx = db.begin().await.unwrap();
for flow in flows {
let status = flow.parse_flow_status();
if status.is_some_and(|s| s.modules.get(0).is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. })))
{
tracing::error!(
"Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
flow.id,
flow.workspace_id
);
// if the flow hasn't started and is a zombie, we can simply restart it
sqlx::query!(
"UPDATE queue SET running = false, started_at = null WHERE id = $1",
flow.id
)
.execute(db)
.await?;
} else {
// if it was started, we can't restart it, so we cancel it
tracing::error!(
"Zombie flow detected: {} in workspace {}. Cancelling it.",
flow.id,
flow.workspace_id
);
let (mut ntx, _) = cancel_job(
"monitor",
Some("Flow cancelled as it was hanging in between 2 steps".to_string()),
flow.id,
flow.workspace_id.as_str(),
tx,
db,
rsmq.clone(),
false,
)
.await?;
// if the flow hasn't started and is a zombie, we can simply restart it
sqlx::query!(
"UPDATE queue SET running = false, started_at = null WHERE id = $1",
flow.id
)
.execute(&mut *ntx)
.await?;
tx = ntx;
}
}
tx.commit().await?;
Ok(())
}
backend/windmill-queue/src/jobs.rs
+16
View File
@@ -564,6 +564,22 @@ pub async fn add_completed_job<
tx = delete_job(tx, &queued_job.workspace_id, job_id).await?;
// tracing::error!("3 {:?}", start.elapsed());
if queued_job.is_flow_step
{
if let Some(parent_job) = queued_job.parent_job {
// persist the flow last progress timestamp to avoid zombie flow jobs
tracing::debug!("Persisting flow last progress timestamp to flow job: {:?}", parent_job);
sqlx::query!(
"UPDATE queue SET last_ping = now() WHERE id = $1 AND workspace_id = $2",
parent_job,
&queued_job.workspace_id
)
.execute(&mut tx)
.await?;
}
}
if !queued_job.is_flow_step
&& queued_job.schedule_path.is_some()
&& queued_job.script_path.is_some()
backend/windmill-worker/Cargo.toml
+1
View File
@@ -15,6 +15,7 @@ enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "dep:
benchmark = ["windmill-queue/benchmark"]
flamegraph = []
parquet = ["windmill-common/parquet"]
flow_testing = []
[dependencies]
windmill-queue.workspace = true
backend/windmill-worker/src/worker.rs
+4 -4
View File
@@ -2373,12 +2373,12 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
};
let update_job_future = if job.is_flow_step || job.is_flow() {
let (flow, job_status_to_update, update_job_future) =
let (flow, job_status_to_update) =
if let Some(parent_job_id) = job.parent_job {
let _ = update_job_future().await;
(parent_job_id, job.id, None)
(parent_job_id, job.id)
} else {
(job.id, Uuid::nil(), Some(update_job_future))
(job.id, Uuid::nil())
};
let wrapped_error = WrappedError { error: err.clone() };
@@ -2421,7 +2421,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
}
}
update_job_future
None
} else {
Some(update_job_future)
};
backend/windmill-worker/src/worker_flow.rs
+66 -104
View File
@@ -9,7 +9,7 @@
use std::collections::hash_map::DefaultHasher;
use std::collections::HashMap;
use std::hash::Hash;
use std::sync::atomic::Ordering;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -73,6 +73,8 @@ pub async fn update_flow_status_after_job_completion<
) -> error::Result<()> {
// this is manual tailrecursion because async_recursion blows up the stack
// todo!();
potentially_crash_for_testing();
let mut rec = update_flow_status_after_job_completion_internal(
db,
client,
@@ -92,6 +94,7 @@ pub async fn update_flow_status_after_job_completion<
)
.await?;
while let Some(nrec) = rec {
potentially_crash_for_testing();
rec = match update_flow_status_after_job_completion_internal(
db,
client,
@@ -578,13 +581,13 @@ pub async fn update_flow_status_after_job_completion_internal<
let done = if !should_continue_flow {
let logs = if flow_job.canceled {
"Flow job canceled".to_string()
"Flow job canceled\n".to_string()
} else if stop_early {
format!("Flow job stopped early because of a stop early predicate returning true")
format!("Flow job stopped early because of a stop early predicate returning true\n")
} else if success {
"Flow job completed with success".to_string()
"Flow job completed with success\n".to_string()
} else {
"Flow job completed with error".to_string()
"Flow job completed with error\n".to_string()
};
#[cfg(feature = "enterprise")]
@@ -1038,30 +1041,6 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.parse_flow_status()
.with_context(|| "Unable to parse flow status")?;
if !flow_job.is_flow_step
&& flow_job.schedule_path.is_some()
&& flow_job.script_path.is_some()
&& status.step == 0
{
let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
match handle_maybe_scheduled_job(
tx,
db,
flow_job.schedule_path.as_ref().unwrap(),
flow_job.script_path.as_ref().unwrap(),
&flow_job.workspace_id,
)
.await
{
Ok(tx) => {
tx.commit().await?;
}
Err(e) => {
tracing::error!("Error during handle_maybe_scheduled_job: {e}");
}
}
}
push_next_flow_job(
flow_job,
@@ -1106,6 +1085,25 @@ pub struct RawArgs {
pub args: Option<Json<HashMap<String, Box<RawValue>>>>,
}
lazy_static::lazy_static! {
static ref CRASH_FORCEFULLY_AT_STEP: Option<usize> = std::env::var("CRASH_FORCEFULLY_AT_STEP")
.ok()
.and_then(|x| x.parse::<usize>().ok());
static ref CRASH_STEP_COUNTER: AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
}
#[inline(always)]
fn potentially_crash_for_testing() {
#[cfg(feature = "flow_testing")]
if let Some(crash_at) = CRASH_FORCEFULLY_AT_STEP.as_ref() {
let counter = CRASH_STEP_COUNTER.fetch_add(1, Ordering::SeqCst);
if &counter == crash_at {
panic!("CRASH#1 - expected crash for testing at step {}", crash_at);
}
}
}
// #[async_recursion]
// #[instrument(level = "trace", skip_all)]
async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
@@ -1160,29 +1158,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
"error sending update flow message to job completed channel: {e}"
))
})?;
// let r;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// if flow.modules.is_empty() {
// r = to_raw_value(&flow_job_args);
// &r
// } else {
// // it has to be an empty for loop event
// serde_json::from_str("[]").unwrap()
// },
// true,
// same_worker_tx,
// worker_dir,
// None,
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
@@ -1229,25 +1205,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
"error sending update flow message to job completed channel: {e}"
))
})?;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// serde_json::from_str(
// "\"not allowed to overlap, scheduling next iteration\"",
// )
// .unwrap(),
// true,
// same_worker_tx,
// worker_dir,
// Some(true),
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
}
@@ -1280,22 +1238,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
))
})?;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// serde_json::from_str("\"stopped early\"").unwrap(),
// true,
// same_worker_tx,
// worker_dir,
// Some(true),
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
}
@@ -1487,6 +1430,14 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE queue
SET last_ping = null
WHERE id = $1 AND last_ping = $2",
flow_job.id,
flow_job.last_ping
).execute(&mut *tx).await?;
tx.commit().await?;
return Ok(());
@@ -1536,23 +1487,6 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
"error sending update flow message to job completed channel: {e}"
))
})?;
// update_flow_status_after_job_completion(
// db,
// client,
// parent_job,
// &flow_job.id,
// &flow_job.workspace_id,
// true,
// &to_raw_value(&result),
// false,
// same_worker_tx.clone(),
// &worker_dir,
// None,
// rsmq,
// worker_name,
// )
// .await?;
return Ok(());
}
}
return Ok(());
@@ -2123,6 +2057,34 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await?;
};
potentially_crash_for_testing();
if !flow_job.is_flow_step
&& status.step == 0
{
if flow_job.schedule_path.is_some()
&& flow_job.script_path.is_some() {
tx = handle_maybe_scheduled_job(
tx,
db,
flow_job.schedule_path.as_ref().unwrap(),
flow_job.script_path.as_ref().unwrap(),
&flow_job.workspace_id,
)
.await?;
}
sqlx::query!(
"UPDATE queue
SET last_ping = null
WHERE id = $1",
flow_job.id
).execute(&mut tx).await?;
}
tx.commit().await?;
tracing::info!(id = %flow_job.id, root_id = %job_root, "all next flow jobs pushed: {uuids:?}");
frontend/src/lib/components/runs/JobLoader.svelte
+1 -1
View File
@@ -59,7 +59,7 @@
function computeJobKinds(jobKindsCat: string | undefined): string {
if (jobKindsCat == 'all') {
return `${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES}`
return `${CompletedJob.job_kind.SCRIPT},${CompletedJob.job_kind.FLOW},${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES},${CompletedJob.job_kind.PREVIEW},${CompletedJob.job_kind.FLOWPREVIEW}`
} else if (jobKindsCat == 'dependencies') {
return `${CompletedJob.job_kind.DEPENDENCIES},${CompletedJob.job_kind.FLOWDEPENDENCIES},${CompletedJob.job_kind.APPDEPENDENCIES}`
} else if (jobKindsCat == 'previews') {