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

v2 list jobs

Browse Source
This commit is contained in:
Ruben Fiszel
2025-02-13 23:47:11 +01:00
parent 2f2a9f3576
commit 70371ef551
2 changed files with 174 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/db.rs
+5 -5
View File
@@ -647,11 +647,11 @@ async fn fix_job_completed_index(db: &DB) -> Result<(), Error> {
.await?;
});
// run_windmill_migration!("v2_jobs_rls", &db, |tx| {
// sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
// .execute(db)
// .await?;
// });
run_windmill_migration!("v2_jobs_rls", &db, |tx| {
sqlx::query!("ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY")
.execute(db)
.await?;
});
Ok(())
}
backend/windmill-api/src/jobs.rs
+169 -136
View File
@@ -1299,27 +1299,30 @@ pub fn filter_list_queue_query(
w_id: &str,
join_outstanding_wait_times: bool,
) -> SqlBuilder {
sqlb.join("v2_job").on_eq("v2_job_queue.id", "v2_job.id");
if join_outstanding_wait_times {
sqlb.left()
.join("outstanding_wait_time")
.on_eq("id", "outstanding_wait_time.job_id");
.on_eq("v2_job.id", "outstanding_wait_time.job_id");
}
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id));
}
if let Some(ps) = &lq.script_path_start {
sqlb.and_where_like_left("script_path", ps);
sqlb.and_where_like_left("runnable_path", ps);
}
if let Some(p) = &lq.script_path_exact {
sqlb.and_where_eq("script_path", "?".bind(p));
sqlb.and_where_eq("runnable_path", "?".bind(p));
}
if let Some(p) = &lq.schedule_path {
sqlb.and_where_eq("schedule_path", "?".bind(p));
sqlb.and_where_eq("trigger", "?".bind(p));
sqlb.and_where_eq("trigger_kind", "schedule");
}
if let Some(h) = &lq.script_hash {
sqlb.and_where_eq("script_hash", "?".bind(h));
sqlb.and_where_eq("runnable_id", "?".bind(h));
}
if let Some(cb) = &lq.created_by {
sqlb.and_where_eq("created_by", "?".bind(cb));
@@ -1340,7 +1343,11 @@ pub fn filter_list_queue_query(
sqlb.and_where_ge("started_at", "?".bind(&dt.to_rfc3339()));
}
if let Some(fs) = &lq.is_flow_step {
sqlb.and_where_eq("is_flow_step", fs);
if *fs {
sqlb.and_where_is_not_null("flow_step_id");
} else {
sqlb.and_where_is_null("flow_step_id");
}
}
if let Some(fs) = &lq.has_null_parent {
if *fs {
@@ -1349,20 +1356,20 @@ pub fn filter_list_queue_query(
}
if let Some(dt) = &lq.created_before {
sqlb.and_where_le("created_at", "?".bind(&dt.to_rfc3339()));
sqlb.and_where_le("v2_job.created_at", "?".bind(&dt.to_rfc3339()));
}
if let Some(dt) = &lq.created_after {
sqlb.and_where_ge("created_at", "?".bind(&dt.to_rfc3339()));
sqlb.and_where_ge("v2_job.created_at", "?".bind(&dt.to_rfc3339()));
}
if let Some(dt) = &lq.created_or_started_after {
let ts = dt.timestamp_millis();
sqlb.and_where(format!("(started_at IS NOT NULL AND started_at >= to_timestamp({} / 1000.0)) OR (started_at IS NULL AND created_at >= to_timestamp({} / 1000.0))", ts, ts));
sqlb.and_where(format!("(started_at IS NOT NULL AND started_at >= to_timestamp({} / 1000.0)) OR (started_at IS NULL AND v2_job.created_at >= to_timestamp({} / 1000.0))", ts, ts));
}
if let Some(dt) = &lq.created_or_started_before {
let ts = dt.timestamp_millis();
sqlb.and_where(format!("(started_at IS NOT NULL AND started_at < to_timestamp({} / 1000.0)) OR (started_at IS NULL AND created_at < to_timestamp({} / 1000.0))", ts, ts));
sqlb.and_where(format!("(started_at IS NOT NULL AND started_at < to_timestamp({} / 1000.0)) OR (started_at IS NULL AND v2_job.created_at < to_timestamp({} / 1000.0))", ts, ts));
}
if let Some(s) = &lq.suspended {
@@ -1375,7 +1382,7 @@ pub fn filter_list_queue_query(
if let Some(jk) = &lq.job_kinds {
sqlb.and_where_in(
"job_kind",
"kind",
&jk.split(',').into_iter().map(quote).collect::<Vec<_>>(),
);
}
@@ -1389,7 +1396,8 @@ pub fn filter_list_queue_query(
}
if lq.is_not_schedule.unwrap_or(false) {
sqlb.and_where("schedule_path IS null");
sqlb.and_where("trigger_kind != 'schedule'")
.or_where("trigger_kind IS NULL");
}
sqlb
@@ -1404,9 +1412,9 @@ pub fn list_queue_jobs_query(
tags: Option<Vec<&str>>,
) -> SqlBuilder {
let (limit, offset) = paginate_without_limits(pagination);
let mut sqlb = SqlBuilder::select_from("v2_as_queue")
let mut sqlb = SqlBuilder::select_from("v2_job_queue")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.order_by("v2_job.created_at", lq.order_desc.unwrap_or(true))
.limit(limit)
.offset(offset)
.clone();
@@ -1451,26 +1459,25 @@ async fn list_queue_jobs(
&w_id,
&lq,
&[
"id",
"running",
"created_by",
"created_at",
"started_at",
"scheduled_for",
"script_hash",
"script_path",
"v2_job.id",
"v2_job_queue.running",
"v2_job.created_by",
"v2_job.created_at",
"v2_job_queue.started_at",
"v2_job_queue.scheduled_for",
"v2_job.runnable_id as script_hash",
"v2_job.runnable_path as script_path",
"null as args",
"job_kind",
"schedule_path",
"permissioned_as",
"is_flow_step",
"language",
"same_worker",
"email",
"suspend",
"tag",
"priority",
"workspace_id",
"v2_job.kind as job_kind",
"CASE WHEN v2_job.trigger_kind = 'schedule' THEN v2_job.trigger END as schedule_path",
"v2_job.permissioned_as",
"v2_job.flow_step_id IS NOT NULL as is_flow_step",
"v2_job.script_lang as language",
"v2_job.permissioned_as_email as email",
"v2_job_queue.suspend",
"v2_job.tag",
"v2_job.priority",
"v2_job.workspace_id",
],
pagination,
false,
@@ -1610,13 +1617,14 @@ async fn list_filtered_uuids(
) -> error::JsonResult<Vec<Uuid>> {
require_admin(authed.is_admin, &authed.username)?;
let mut sqlb = SqlBuilder::select_from("v2_as_queue")
.fields(&["id"])
let mut sqlb = SqlBuilder::select_from("v2_job_queue")
.fields(&["v2_job_queue.id"])
.clone();
sqlb = join_concurrency_key(lq.concurrency_key.as_ref(), sqlb);
sqlb.and_where_is_null("schedule_path");
sqlb.and_where_ne("v2_job.trigger_kind", "'schedule'")
.or_where_is_null("v2_job.trigger_kind");
if let Some(tags) = get_scope_tags(&authed) {
sqlb.and_where_in("tag", &tags.iter().map(|x| quote(x)).collect::<Vec<_>>());
@@ -1630,7 +1638,7 @@ async fn list_filtered_uuids(
Ok(Json(jobs))
}
#[derive(Serialize, Debug, FromRow)]
#[derive(Serialize)]
struct QueueStats {
database_length: i64,
suspended: Option<i64>,
@@ -1671,23 +1679,28 @@ async fn count_completed_jobs_detail(
Path(w_id): Path<String>,
Query(query): Query<CountCompletedJobsQuery>,
) -> error::JsonResult<i64> {
let mut sqlb = SqlBuilder::select_from("v2_as_completed_job");
let mut sqlb = SqlBuilder::select_from("v2_job_completed");
//FOR RLS
sqlb.join("v2_job USING (id)");
sqlb.field("COUNT(*) as count");
if !query.all_workspaces.unwrap_or(false) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id));
}
if let Some(after_s_ago) = query.completed_after_s_ago {
let after = Utc::now() - chrono::Duration::seconds(after_s_ago);
sqlb.and_where_gt(
"started_at + duration_ms / 1000 * interval '1 second'",
"?".bind(&after.to_rfc3339()),
);
sqlb.and_where_gt("ended_at", "?".bind(&after.to_rfc3339()));
}
if let Some(success) = query.success {
sqlb.and_where_eq("success", "?".bind(&success));
if success {
sqlb.and_where_eq("status", "'success'")
.or_where_eq("status", "'skipped'");
} else {
sqlb.and_where_ne("status", "'success'")
.and_where_ne("status", "'skipped'");
}
}
if let Some(tags) = query.tags {
@@ -2656,77 +2669,78 @@ pub struct UnifiedJob {
const CJ_FIELDS: &[&str] = &[
"'CompletedJob' as typ",
"id",
"workspace_id",
"parent_job",
"created_by",
"created_at",
"started_at",
"v2_job.id",
"v2_job.workspace_id",
"v2_job.parent_job",
"v2_job.created_by",
"v2_job.created_at",
"v2_job_completed.started_at",
"null as scheduled_for",
"null as running",
"script_hash",
"script_path",
"v2_job.runnable_id as script_hash",
"v2_job.runnable_path as script_path",
"null as args",
"duration_ms",
"success",
"deleted",
"canceled",
"canceled_by",
"job_kind",
"schedule_path",
"permissioned_as",
"is_flow_step",
"language",
"is_skipped",
"email",
"visible_to_owner",
"v2_job_completed.duration_ms",
"v2_job_completed.status = 'success' as success",
"false as deleted",
"v2_job_completed.status = 'canceled' as canceled",
"v2_job_completed.canceled_by",
"v2_job.kind as job_kind",
"CASE WHEN v2_job.trigger_kind = 'schedule' THEN v2_job.trigger END as schedule_path",
"v2_job.permissioned_as",
"v2_job.flow_step_id IS NOT NULL as is_flow_step",
"v2_job.script_lang as language",
"v2_job_completed.status = 'skipped' as is_skipped",
"v2_job.permissioned_as_email as email",
"v2_job.visible_to_owner",
"null as suspend",
"mem_peak",
"tag",
"v2_job_completed.memory_peak as mem_peak",
"v2_job.tag",
"null as concurrent_limit",
"null as concurrency_time_window_s",
"priority",
"result->'wm_labels' as labels",
"v2_job.priority",
"v2_job_completed.result->'wm_labels' as labels",
"self_wait_time_ms",
"aggregate_wait_time_ms",
"preprocessed",
"v2_job.preprocessed",
];
const QJ_FIELDS: &[&str] = &[
"'QueuedJob' as typ",
"id",
"workspace_id",
"parent_job",
"created_by",
"created_at",
"started_at",
"scheduled_for",
"running",
"script_hash",
"script_path",
"v2_job.id",
"v2_job.workspace_id",
"v2_job.parent_job",
"v2_job.created_by",
"v2_job.created_at",
"v2_job_queue.started_at",
"v2_job_queue.scheduled_for",
"v2_job_queue.running",
"v2_job.runnable_id as script_hash",
"v2_job.runnable_path as script_path",
"null as args",
"null as duration_ms",
"null as success",
"false as deleted",
"canceled",
"canceled_by",
"job_kind",
"schedule_path",
"permissioned_as",
"is_flow_step",
"language",
"v2_job_queue.canceled_by IS NOT NULL as canceled",
"v2_job_queue.canceled_by",
"v2_job.kind as job_kind",
"CASE WHEN v2_job.trigger_kind = 'schedule' THEN v2_job.trigger END as schedule_path",
"v2_job.permissioned_as",
"v2_job.flow_step_id IS NOT NULL as is_flow_step",
"v2_job.script_lang as language",
"false as is_skipped",
"email",
"visible_to_owner",
"suspend",
"mem_peak",
"tag",
"concurrent_limit",
"concurrency_time_window_s",
"priority",
"v2_job.permissioned_as_email as email",
"v2_job.visible_to_owner",
"v2_job_queue.suspend",
"null as mem_peak",
"v2_job.tag",
"v2_job.concurrent_limit",
"v2_job.concurrency_time_window_s",
"v2_job.priority",
"null as labels",
"self_wait_time_ms",
"aggregate_wait_time_ms",
"preprocessed",
"v2_job.preprocessed",
];
impl UnifiedJob {
@@ -5258,10 +5272,13 @@ pub fn filter_list_completed_query(
w_id: &str,
join_outstanding_wait_times: bool,
) -> SqlBuilder {
sqlb.join("v2_job")
.on_eq("v2_job_completed.id", "v2_job.id");
if join_outstanding_wait_times {
sqlb.left()
.join("outstanding_wait_time")
.on_eq("id", "outstanding_wait_time.job_id");
.on_eq("v2_job.id", "outstanding_wait_time.job_id");
}
if let Some(label) = &lq.label {
@@ -5272,21 +5289,22 @@ pub fn filter_list_completed_query(
}
if w_id != "admins" || !lq.all_workspaces.is_some_and(|x| x) {
sqlb.and_where_eq("workspace_id", "?".bind(&w_id));
sqlb.and_where_eq("v2_job.workspace_id", "?".bind(&w_id));
}
if let Some(p) = &lq.schedule_path {
sqlb.and_where_eq("schedule_path", "?".bind(p));
sqlb.and_where_eq("trigger", "?".bind(p));
sqlb.and_where_eq("trigger_kind", "'schedule'");
}
if let Some(ps) = &lq.script_path_start {
sqlb.and_where_like_left("script_path", ps);
sqlb.and_where_like_left("runnable_path", ps);
}
if let Some(p) = &lq.script_path_exact {
sqlb.and_where_eq("script_path", "?".bind(p));
sqlb.and_where_eq("runnable_path", "?".bind(p));
}
if let Some(h) = &lq.script_hash {
sqlb.and_where_eq("script_hash", "?".bind(h));
sqlb.and_where_eq("runnable_id", "?".bind(h));
}
if let Some(t) = &lq.tag {
sqlb.and_where_eq("tag", "?".bind(t));
@@ -5295,7 +5313,13 @@ pub fn filter_list_completed_query(
sqlb.and_where_eq("created_by", "?".bind(cb));
}
if let Some(r) = &lq.success {
sqlb.and_where_eq("success", r);
if *r {
sqlb.and_where_eq("status", "'success'")
.or_where_eq("status", "'skipped'");
} else {
sqlb.and_where_eq("status", "'failure'")
.or_where_eq("status", "'canceled'");
}
}
if let Some(pj) = &lq.parent_job {
sqlb.and_where_eq("parent_job", "?".bind(pj));
@@ -5326,10 +5350,18 @@ pub fn filter_list_completed_query(
}
if let Some(sk) = &lq.is_skipped {
sqlb.and_where_eq("is_skipped", sk);
if *sk {
sqlb.and_where_eq("status", "'skipped'");
} else {
sqlb.and_where_ne("status", "'skipped'");
}
}
if let Some(fs) = &lq.is_flow_step {
sqlb.and_where_eq("is_flow_step", fs);
if *fs {
sqlb.and_where_is_not_null("flow_step_id");
} else {
sqlb.and_where_is_null("flow_step_id");
}
}
if let Some(fs) = &lq.has_null_parent {
if *fs {
@@ -5338,7 +5370,7 @@ pub fn filter_list_completed_query(
}
if let Some(jk) = &lq.job_kinds {
sqlb.and_where_in(
"job_kind",
"kind",
&jk.split(',').into_iter().map(quote).collect::<Vec<_>>(),
);
}
@@ -5352,7 +5384,8 @@ pub fn filter_list_completed_query(
}
if lq.is_not_schedule.unwrap_or(false) {
sqlb.and_where("schedule_path IS null");
sqlb.and_where("trigger_kind != 'schedule'")
.or_where("trigger_kind IS NULL");
}
sqlb
@@ -5367,9 +5400,9 @@ pub fn list_completed_jobs_query(
join_outstanding_wait_times: bool,
tags: Option<Vec<&str>>,
) -> SqlBuilder {
let mut sqlb = SqlBuilder::select_from("v2_as_completed_job")
let mut sqlb = SqlBuilder::select_from("v2_job_completed")
.fields(fields)
.order_by("created_at", lq.order_desc.unwrap_or(true))
.order_by("v2_job.created_at", lq.order_desc.unwrap_or(true))
.offset(offset)
.limit(per_page)
.clone();
@@ -5432,35 +5465,35 @@ async fn list_completed_jobs(
offset,
&lq,
&[
"id",
"workspace_id",
"parent_job",
"created_by",
"created_at",
"started_at",
"duration_ms",
"success",
"script_hash",
"script_path",
"deleted",
"canceled",
"canceled_by",
"canceled_reason",
"job_kind",
"schedule_path",
"permissioned_as",
"v2_job.id",
"v2_job.workspace_id",
"v2_job.parent_job",
"v2_job.created_by",
"v2_job.created_at",
"v2_job_completed.started_at",
"v2_job_completed.duration_ms",
"v2_job_completed.status = 'success' as success",
"v2_job.runnable_id as script_hash",
"v2_job.runnable_path as script_path",
"false as deleted",
"v2_job_completed.status = 'canceled' as canceled",
"v2_job_completed.canceled_by",
"v2_job_completed.canceled_reason",
"v2_job.kind as job_kind",
"CASE WHEN v2_job.trigger_kind = 'schedule' THEN v2_job.trigger END as schedule_path",
"v2_job.permissioned_as",
"null as raw_code",
"null as flow_status",
"null as raw_flow",
"is_flow_step",
"language",
"is_skipped",
"email",
"visible_to_owner",
"mem_peak",
"tag",
"priority",
"result->'wm_labels' as labels",
"v2_job.flow_step_id IS NOT NULL as is_flow_step",
"v2_job.script_lang as language",
"v2_job_completed.status = 'skipped' as is_skipped",
"v2_job.permissioned_as_email as email",
"v2_job.visible_to_owner",
"v2_job_completed.memory_peak as mem_peak",
"v2_job.tag",
"v2_job.priority",
"v2_job_completed.result->'wm_labels' as labels",
"'CompletedJob' as type",
],
false,