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

fix: improve runs page performance through pg indices

Browse Source
This commit is contained in:
Ruben Fiszel
2024-07-27 12:16:13 +02:00
parent 99981a5874
commit 135f57eb5a
2 changed files with 141 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/db.rs
+58 -6
View File
@@ -249,14 +249,55 @@ async fn fix_flow_versioning_migration(
}
async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
// let has_done_migration = sqlx::query_scalar!(
// "SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"
// )
// .fetch_one(db)
// .await?
// .unwrap_or(false);
// if !has_done_migration {
// tracing::info!("Applying fix_job_completed_index migration");
// let mut tx = db.begin().await?;
// let mut r = false;
// while !r {
// r = sqlx::query_scalar!("SELECT pg_try_advisory_lock(4242)")
// .fetch_one(&mut *tx)
// .await
// .map_err(|e| {
// tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}");
// sqlx::migrate::MigrateError::Execute(e)
// })?
// .unwrap_or(false);
// if !r {
// tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
// tokio::time::sleep(std::time::Duration::from_secs(5)).await;
// }
// }
// // sqlx::query(
// // "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)"
// // ).execute(db).await?;
// sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
// .execute(db)
// .await?;
// sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING")
// .execute(&mut *tx)
// .await?;
// let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
// .execute(&mut *tx)
// .await?;
// tx.commit().await?;
// }
let has_done_migration = sqlx::query_scalar!(
"SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index')"
"SELECT EXISTS(SELECT name FROM windmill_migrations WHERE name = 'fix_job_completed_index_2')"
)
.fetch_one(db)
.await?
.unwrap_or(false);
if !has_done_migration {
tracing::info!("Applying fix_job_completed_index migration");
tracing::info!("Applying fix_job_completed_index_2 migration");
let mut tx = db.begin().await?;
let mut r = false;
while !r {
@@ -264,30 +305,41 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
.fetch_one(&mut *tx)
.await
.map_err(|e| {
tracing::error!("Error acquiring fix_job_completed_index lock: {e:#}");
tracing::error!("Error acquiring fix_job_completed_index_2 lock: {e:#}");
sqlx::migrate::MigrateError::Execute(e)
})?
.unwrap_or(false);
if !r {
tracing::info!("PG fix_job_completed_index_migration lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
tracing::info!("PG fix_job_completed_index_migration_2 lock already acquired by another server or worker, retrying in 5s. (look for the advisory lock in pg_lock with granted = true)");
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}
}
sqlx::query(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new ON completed_job (workspace_id, job_kind, is_skipped, is_flow_step, created_at DESC, started_at DESC)"
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_created_at_new_2 ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, created_at DESC)"
).execute(db).await?;
sqlx::query(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_completed_job_workspace_id_started_at_new ON completed_job (workspace_id, job_kind, success, is_skipped, is_flow_step, started_at DESC)"
).execute(db).await?;
sqlx::query("DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at")
.execute(db)
.await?;
sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index') ON CONFLICT DO NOTHING")
sqlx::query(
"DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new",
)
.execute(db)
.await?;
sqlx::query!("INSERT INTO windmill_migrations (name) VALUES ('fix_job_completed_index_2') ON CONFLICT DO NOTHING")
.execute(&mut *tx)
.await?;
let _ = sqlx::query("SELECT pg_advisory_unlock(4242)")
.execute(&mut *tx)
.await?;
tx.commit().await?;
tracing::info!("Finished applying fix_job_completed_index_2 migration");
}
Ok(())
frontend/src/lib/components/runs/JobLoader.svelte
+83 -54
View File
@@ -108,37 +108,49 @@
}
}
let loadingFetch = false
async function fetchJobs(
startedBefore: string | undefined,
startedAfter: string | undefined,
startedAfterCompletedJobs: string | undefined
): Promise<Job[]> {
return JobService.listJobs({
workspace: $workspaceStore!,
createdOrStartedBefore: startedBefore,
createdOrStartedAfter: startedAfter,
createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs,
schedulePath,
scriptPathExact: path === null || path === '' ? undefined : path,
createdBy: user === null || user === '' ? undefined : user,
scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`,
jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
isNotSchedule: showSchedules == false ? true : undefined,
scheduledForBeforeNow: showFutureJobs == false ? true : undefined,
args:
argFilter && argFilter != '{}' && argFilter != '' && argError == '' ? argFilter : undefined,
result:
resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == ''
? resultFilter
: undefined,
allWorkspaces: allWorkspaces ? true : undefined,
perPage
})
loadingFetch = true
try {
return JobService.listJobs({
workspace: $workspaceStore!,
createdOrStartedBefore: startedBefore,
createdOrStartedAfter: startedAfter,
createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs,
schedulePath,
scriptPathExact: path === null || path === '' ? undefined : path,
createdBy: user === null || user === '' ? undefined : user,
scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`,
jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
isNotSchedule: showSchedules == false ? true : undefined,
scheduledForBeforeNow: showFutureJobs == false ? true : undefined,
args:
argFilter && argFilter != '{}' && argFilter != '' && argError == ''
? argFilter
: undefined,
result:
resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == ''
? resultFilter
: undefined,
allWorkspaces: allWorkspaces ? true : undefined,
perPage
})
} catch (e) {
sendUserToast('There was an issue loading jobs, see browser console for more details', true)
console.error(e)
return []
} finally {
loadingFetch = false
}
}
async function fetchExtendedJobs(
@@ -147,34 +159,48 @@
startedAfter: string | undefined,
startedAfterCompletedJobs: string | undefined
): Promise<ExtendedJobs> {
return ConcurrencyGroupsService.listExtendedJobs({
rowLimit: 1000,
concurrencyKey: concurrencyKey == null || concurrencyKey == '' ? undefined : concurrencyKey,
workspace: $workspaceStore!,
createdOrStartedBefore: startedBefore,
createdOrStartedAfter: startedAfter,
createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs,
schedulePath,
scriptPathExact: path === null || path === '' ? undefined : path,
createdBy: user === null || user === '' ? undefined : user,
scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`,
jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
isNotSchedule: showSchedules == false ? true : undefined,
scheduledForBeforeNow: showFutureJobs == false ? true : undefined,
args:
argFilter && argFilter != '{}' && argFilter != '' && argError == '' ? argFilter : undefined,
result:
resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == ''
? resultFilter
: undefined,
allWorkspaces: allWorkspaces ? true : undefined,
perPage
})
loadingFetch = true
try {
return ConcurrencyGroupsService.listExtendedJobs({
rowLimit: 1000,
concurrencyKey: concurrencyKey == null || concurrencyKey == '' ? undefined : concurrencyKey,
workspace: $workspaceStore!,
createdOrStartedBefore: startedBefore,
createdOrStartedAfter: startedAfter,
createdOrStartedAfterCompletedJobs: startedAfterCompletedJobs,
schedulePath,
scriptPathExact: path === null || path === '' ? undefined : path,
createdBy: user === null || user === '' ? undefined : user,
scriptPathStart: folder === null || folder === '' ? undefined : `f/${folder}/`,
jobKinds,
success: success == 'success' ? true : success == 'failure' ? false : undefined,
running: success == 'running' ? true : undefined,
isSkipped: isSkipped ? undefined : false,
isFlowStep: jobKindsCat != 'all' ? false : undefined,
label: label === null || label === '' ? undefined : label,
isNotSchedule: showSchedules == false ? true : undefined,
scheduledForBeforeNow: showFutureJobs == false ? true : undefined,
args:
argFilter && argFilter != '{}' && argFilter != '' && argError == ''
? argFilter
: undefined,
result:
resultFilter && resultFilter != '{}' && resultFilter != '' && resultError == ''
? resultFilter
: undefined,
allWorkspaces: allWorkspaces ? true : undefined,
perPage
})
} catch (e) {
sendUserToast('There was an issue loading jobs, see browser console for more details', true)
console.error(e)
return {
jobs: [],
obscured_jobs: []
}
} finally {
loadingFetch = false
}
}
export async function loadJobs(
@@ -256,6 +282,9 @@
}
async function syncer() {
if (loadingFetch) {
return
}
if (sync) {
if (syncQueuedRunsCount) {
getCount()