diff --git a/backend/migrations/20230724145302_improve_pull_with_index.down.sql b/backend/migrations/20230724145302_improve_pull_with_index.down.sql new file mode 100644 index 0000000000..556aa24dff --- /dev/null +++ b/backend/migrations/20230724145302_improve_pull_with_index.down.sql @@ -0,0 +1,3 @@ +-- Add down migration script here + +DROP INDEX index_queue_on_main_filter; \ No newline at end of file diff --git a/backend/migrations/20230724145302_improve_pull_with_index.up.sql b/backend/migrations/20230724145302_improve_pull_with_index.up.sql new file mode 100644 index 0000000000..0da0a538a3 --- /dev/null +++ b/backend/migrations/20230724145302_improve_pull_with_index.up.sql @@ -0,0 +1 @@ +-- Add up migration script here diff --git a/backend/src/main.rs b/backend/src/main.rs index 37c401f58b..8111a1ac05 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -160,8 +160,6 @@ Windmill Community Edition {GIT_VERSION} "PIP_LOCAL_DEPENDENCIES", "ADDITIONAL_PYTHON_PATHS", "INCLUDE_HEADERS", - "WHITELIST_WORKSPACES", - "BLACKLIST_WORKSPACES", "INSTANCE_EVENTS_WEBHOOK", "CLOUD_HOSTED", "GLOBAL_CACHE_INTERVAL", @@ -181,6 +179,12 @@ Windmill Community Edition {GIT_VERSION} "GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE", ]); + if std::env::var("WHITELIST_WORKSPACES").is_ok() + || std::env::var("BLACKLIST_WORKSPACES").is_ok() + { + panic!("WHITELIST_WORKSPACES and BLACKLIST_WORKSPACES have been removed, please use Worker Groups instead"); + } + tracing::info!("Loading OAuth providers...: {:#?}", *OAUTH_CLIENTS); if let Some(ref smtp) = *SMTP_CLIENT { tracing::info!("Smtp client defined. Testing connection..."); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 5b85adca53..03d7426b2f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -82,8 +82,30 @@ lazy_static::lazy_static! { "hub".to_string(), "other".to_string()]); - pub static ref ACCEPTED_TAGS_FILTER: String = format!(" AND ({})", - ACCEPTED_TAGS.clone().into_iter().map(|x| format!("(tag = '{x}')")).join(" OR ")); + + pub static ref PULL_QUERY: String = format!( + "UPDATE queue + SET running = true + , started_at = coalesce(started_at, now()) + , last_ping = now() + , suspend_until = null + WHERE id = ( + SELECT id + FROM queue + WHERE ((running = false + AND scheduled_for <= now()) + OR (suspend_until IS NOT NULL + AND ( suspend <= 0 + OR suspend_until <= now()))) + {} + ORDER BY scheduled_for, created_at + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) LIMIT 1 + RETURNING *", + format!(" AND ({})", + ACCEPTED_TAGS.clone().into_iter().map(|x| format!("(tag = '{x}')")).join(" OR ")) + ); // When compiled in 'benchmark' mode, this flags is exposed via the /workers/toggle endpoint // and make it possible to disable to current active workers (such that they don't pull any) @@ -660,49 +682,15 @@ async fn handle_on_failure<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c pub async fn pull( db: &Pool, - whitelist_workspaces: Option>, - blacklist_workspaces: Option>, rsmq: Option, ) -> windmill_common::error::Result> { - let mut workspaces_filter = String::new(); - if let Some(whitelist) = whitelist_workspaces { - workspaces_filter.push_str(&format!( - " AND workspace_id IN ({})", - whitelist - .into_iter() - .map(|x| format!("'{x}'")) - .collect::>() - .join(",") - )); - if let Some(_rsmq) = rsmq { - todo!("REDIS: Implement workspace filters for redis"); - } - } - if let Some(blacklist) = blacklist_workspaces { - workspaces_filter.push_str(&format!( - " AND workspace_id NOT IN ({})", - blacklist - .into_iter() - .map(|x| format!("'{x}'")) - .collect::>() - .join(",") - )); - if let Some(_rsmq) = rsmq { - todo!("REDIS: Implement workspace filters for redis"); - } - } - // let rs = rd_string(2); // let instant = Instant::now(); loop { let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.clone().begin().await?).into(); - let (job, mut tx) = pull_single_job_and_mark_as_running_no_concurrency_limit( - tx, - workspaces_filter.as_str(), - rsmq.clone(), - ) - .await?; + let (job, mut tx) = + pull_single_job_and_mark_as_running_no_concurrency_limit(tx, rsmq.clone()).await?; if job.is_none() { return Ok(None); @@ -842,7 +830,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< R: rsmq_async::RsmqConnection + Send + Clone, >( mut tx: QueueTransaction<'c, R>, - workspaces_filter: &str, rsmq: Option, ) -> windmill_common::error::Result<(Option, QueueTransaction<'c, R>)> { let job: Option = if let Some(mut rsmq) = rsmq { @@ -876,7 +863,6 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< None } } else { - let accepted_tags_filter = &*ACCEPTED_TAGS_FILTER; /* Jobs can be started if they: * - haven't been started before, * running = false @@ -884,30 +870,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit< * suspend_until is non-null * and suspend = 0 when the resume messages are received * or suspend_until <= now() if it has timed out */ - sqlx::query_as::<_, QueuedJob>(&format!( - "UPDATE queue - SET running = true - , started_at = coalesce(started_at, now()) - , last_ping = now() - , suspend_until = null - WHERE id = ( - SELECT id - FROM queue - WHERE ((running = false - AND scheduled_for <= now()) - OR (suspend_until IS NOT NULL - AND ( suspend <= 0 - OR suspend_until <= now()))) - {workspaces_filter} - {accepted_tags_filter} - ORDER BY scheduled_for - FOR UPDATE SKIP LOCKED - LIMIT 1 - ) - RETURNING *" - )) - .fetch_optional(&mut tx) - .await? + sqlx::query_as::<_, QueuedJob>(&PULL_QUERY) + .fetch_optional(&mut tx) + .await? }; Ok((job, tx)) } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 5073a6d12e..8c967bdd74 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -212,13 +212,6 @@ lazy_static::lazy_static! { .ok() .map(|x| x.split(',').map(|x| (x.to_string(), std::env::var(x).unwrap_or("".to_string()))).collect()); - static ref WHITELIST_WORKSPACES: Option> = std::env::var("WHITELIST_WORKSPACES") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).collect()); - static ref BLACKLIST_WORKSPACES: Option> = std::env::var("BLACKLIST_WORKSPACES") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).collect()); - pub static ref TAR_CACHE_RATE: i32 = std::env::var("TAR_CACHE_RATE") .ok() .and_then(|x| x.parse::().ok()) @@ -610,7 +603,7 @@ pub async fn run_worker { timer.map(|timer| { let duration_pull_s = timer.stop_and_record(); diff --git a/benchmarks/worker.ts b/benchmarks/worker.ts index f800d30d3d..54d4890b54 100644 --- a/benchmarks/worker.ts +++ b/benchmarks/worker.ts @@ -272,16 +272,14 @@ while (cont) { language: "postgresql", args: { query: "SELECT email FROM usr", - database_url: - "postgres://postgres:changeme@localhost:5432/windmill", + database_url: "postgres://postgres:changeme@localhost:5432/windmill", }, }; } else { payload = { path: "denosimple", language: api.Preview.language.DENO, - content: - 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', + content: 'export function main(){ return Deno.env.get("WM_JOB_ID"); }', args: {}, }; } @@ -300,7 +298,6 @@ while (cont) { clearInterval(updateStatusInterval); - const end_time = Date.now() + complete_timeout; let incorrect_results = 0;