diff --git a/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json b/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json new file mode 100644 index 0000000000..5fa8d88b2e --- /dev/null +++ b/backend/.sqlx/query-0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3.json @@ -0,0 +1,19 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status,\n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n $4\n ),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5),\n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6)\n ELSE\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4)\n END\n WHERE id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "0088b5bf5cf1c7e47f18c3d05cce09f3462c8e39cb3291f0d497742ad4763fa3" +} diff --git a/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json b/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json deleted file mode 100644 index 39b7179e5c..0000000000 --- a/backend/.sqlx/query-1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n flow_status,\n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT],\n $4\n )\n WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "1060c503cf8d4bb5cef9720c162b8192924b4a938d249fae92624cd55e44f488" -} diff --git a/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json b/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json new file mode 100644 index 0000000000..1734a545fe --- /dev/null +++ b/backend/.sqlx/query-140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT flow_status->'modules'->$2::int->'flow_jobs_success' as \"flow_jobs_success: Json>>\", flow_status->'modules'->$2::int->'flow_jobs_duration' as \"flow_jobs_duration: Json\"\n FROM v2_job_status WHERE id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "flow_jobs_success: Json>>", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "flow_jobs_duration: Json", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Int4" + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "140e77db6b38574c62f35d23b876e749d2c43af837f2b3f3edbf49f979e44082" +} diff --git a/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json b/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json new file mode 100644 index 0000000000..9deb40619a --- /dev/null +++ b/backend/.sqlx/query-2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE \n WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n )\n ELSE\n JSONB_SET(JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n ) END\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "2d0df658c31d45592dca7add8c69d06da7d81e7177a324e3f3f9af1076b3ab66" +} diff --git a/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json b/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json new file mode 100644 index 0000000000..0d8d788602 --- /dev/null +++ b/backend/.sqlx/query-8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "started_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "UuidArray" + ] + }, + "nullable": [ + false, + true + ] + }, + "hash": "8167af4650c99dc66de483d72a675a7b57946b430ad0839685797a10ac1adfd8" +} diff --git a/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json b/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json deleted file mode 100644 index 0f6659264a..0000000000 --- a/backend/.sqlx/query-96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [ - null - ] - }, - "hash": "96c0e34708bbba29db162e7289a942addd4581dddc88663b6c2cbae87ec205fc" -} diff --git a/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json b/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json new file mode 100644 index 0000000000..1315df082f --- /dev/null +++ b/backend/.sqlx/query-a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f.json @@ -0,0 +1,27 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE v2_job_status SET flow_status = \n CASE \n WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN\n JSONB_SET(\n JSONB_SET(JSONB_SET(JSONB_SET(\n flow_status, \n ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), \n ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n ELSE\n JSONB_SET(JSONB_SET(\n flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'branchall', 'branch'],\n ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb\n )\n END\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "int4", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Int4", + "Uuid", + "Text", + "Jsonb", + "Jsonb", + "Jsonb" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a49e59c814d440bfed7d6bfc2755518225b1243a9d39ce659b95637c33d0954f" +} diff --git a/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json b/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json deleted file mode 100644 index 09f24968f3..0000000000 --- a/backend/.sqlx/query-d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE v2_job_status SET\n flow_status = JSONB_SET(\n JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4),\n ARRAY['modules', $1::TEXT, 'iterator', 'index'],\n ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb\n )\n WHERE id = $2\n RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "int4", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Int4", - "Uuid", - "Text", - "Jsonb" - ] - }, - "nullable": [ - null - ] - }, - "hash": "d6db1103fee4bad6831656d77a28254c9a832b4f660ec755f4fa14f6f7bb3353" -} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c83b1ca3ab..40e3ab25e7 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8024,6 +8024,34 @@ paths: application/json: schema: {} + /w/{workspace}/jobs_u/queue/get_started_at_by_ids: + post: + summary: get started at by ids + operationId: getStartedAtByIds + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: ids + required: true + content: + application/json: + schema: + type: array + items: + type: string + responses: + "200": + description: started at by ids + content: + application/json: + schema: + type: array + items: + type: string + format: date-time + /w/{workspace}/jobs_u/getupdate/{id}: get: summary: get job updates @@ -18249,7 +18277,7 @@ components: allow: type: string # comma separated permissions : "read,write,delete,list" required: ["pattern", "allow"] - + GitRepositorySettings: type: object properties: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index b00a08cdb5..1b75070217 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -312,6 +312,7 @@ pub fn workspace_unauthed_service() -> Router { get(get_completed_job_logs_tail), ) .route("/get_args/:id", get(get_args)) + .route("/queue/get_started_at_by_ids", post(get_started_at_by_ids)) .route("/get_flow_debug_info/:id", get(get_flow_job_debug_info)) .route("/completed/get/:id", get(get_completed_job)) .route("/completed/get_result/:id", get(get_completed_job_result)) @@ -1663,6 +1664,32 @@ async fn get_args( } } +async fn get_started_at_by_ids( + Extension(db): Extension, + Json(mut ids): Json>, +) -> JsonResult>>> { + ids.truncate(100); + + let started_at = sqlx::query!( + "SELECT id, started_at FROM v2_job_queue WHERE id = ANY($1)", + ids.as_slice() + ) + .fetch_all(&db) + .await?; + + let as_map = started_at + .iter() + .map(|x| (x.id, x.started_at)) + .collect::>(); + + let mut r = Vec::new(); + for id in ids { + r.push(as_map.get(&id).map(|x| x.clone()).unwrap_or_default()); + } + + Ok(Json(r)) +} + #[derive(Debug, sqlx::FromRow, Serialize)] pub struct ListableCompletedJob { pub r#type: String, @@ -6868,6 +6895,7 @@ fn start_job_update_sse_stream( update.log_offset = None; } } + if let Some(new_stream_offset) = update.stream_offset { if new_stream_offset != stream_offset.unwrap_or(0) { stream_offset = Some(new_stream_offset); diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 8b16e6f427..8dde5606fd 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/backend/windmill-common/src/flow_status.rs @@ -113,6 +113,50 @@ pub struct FlowCleanupModule { pub flow_jobs_to_clean: Vec, } +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct FlowJobsDuration { + pub started_at: Vec>>, + pub duration_ms: Vec>, +} + +impl FlowJobsDuration { + pub fn set(&mut self, position: Option, value: &Option) { + if let Some(position) = position { + if position >= self.started_at.len() + || position >= self.duration_ms.len() + || value.is_none() + { + return; + } + let value = value.clone().unwrap(); + self.started_at[position] = Some(value.started_at); + self.duration_ms[position] = Some(value.duration_ms); + } + } + + pub fn push(&mut self, value: &Option) { + self.started_at.push(value.as_ref().map(|x| x.started_at)); + self.duration_ms.push(value.as_ref().map(|x| x.duration_ms)); + } + + pub fn new(n: usize) -> Self { + Self { started_at: vec![None; n], duration_ms: vec![None; n] } + } +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct FlowJobDuration { + pub started_at: chrono::DateTime, + pub duration_ms: i64, +} + +impl FlowJobsDuration { + pub fn truncate(&mut self, n: usize) { + self.started_at.truncate(n); + self.duration_ms.truncate(n); + } +} + #[derive(Deserialize)] struct UntaggedFlowStatusModule { #[serde(rename = "type")] @@ -124,6 +168,7 @@ struct UntaggedFlowStatusModule { iterator: Option, flow_jobs: Option>, flow_jobs_success: Option>>, + flow_jobs_duration: Option, branch_chosen: Option, branchall: Option, parallel: Option, @@ -169,6 +214,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(skip_serializing_if = "Option::is_none")] branchall: Option, @@ -189,6 +236,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(default)] #[serde(skip_serializing_if = "Vec::is_empty")] @@ -209,6 +258,8 @@ pub enum FlowStatusModule { #[serde(skip_serializing_if = "Option::is_none")] flow_jobs_success: Option>>, #[serde(skip_serializing_if = "Option::is_none")] + flow_jobs_duration: Option, + #[serde(skip_serializing_if = "Option::is_none")] branch_chosen: Option, #[serde(skip_serializing_if = "Vec::is_empty")] failed_retries: Vec, @@ -262,6 +313,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { iterator: untagged.iterator, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, branchall: untagged.branchall, parallel: untagged.parallel.unwrap_or(false), @@ -279,6 +331,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { .ok_or_else(|| serde::de::Error::missing_field("job"))?, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, approvers: untagged.approvers.unwrap_or_default(), failed_retries: untagged.failed_retries.unwrap_or_default(), @@ -295,6 +348,7 @@ impl<'de> Deserialize<'de> for FlowStatusModule { .ok_or_else(|| serde::de::Error::missing_field("job"))?, flow_jobs: untagged.flow_jobs, flow_jobs_success: untagged.flow_jobs_success, + flow_jobs_duration: untagged.flow_jobs_duration, branch_chosen: untagged.branch_chosen, failed_retries: untagged.failed_retries.unwrap_or_default(), agent_actions: untagged.agent_actions, @@ -360,6 +414,15 @@ impl FlowStatusModule { } } + pub fn flow_jobs_duration(&self) -> Option { + match self { + FlowStatusModule::InProgress { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + FlowStatusModule::Success { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + FlowStatusModule::Failure { flow_jobs_duration, .. } => flow_jobs_duration.clone(), + _ => None, + } + } + pub fn job_result(&self) -> Option { self.flow_jobs() .map(JobResult::ListJob) diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7a7815dc6c..7c6a3b3dcd 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -758,7 +758,7 @@ pub async fn add_completed_job( canceled_by: Option, flow_is_done: bool, duration: Option, -) -> Result { +) -> Result<(Uuid, i64), Error> { // tracing::error!("Start"); // let start = tokio::time::Instant::now(); @@ -770,7 +770,7 @@ pub async fn add_completed_job( } let result_columns = result_columns.as_ref(); - let (opt_uuid, _duration, _skip_downstream_error_handlers) = (|| { + let (opt_uuid, duration, _skip_downstream_error_handlers) = (|| { commit_completed_job( db, queued_job, @@ -799,11 +799,11 @@ pub async fn add_completed_job( // if scheduling next job failed, return the job_id early to ensure the job get retried after a timeout if let Some(job_id) = opt_uuid { - return Ok(job_id); + return Ok((job_id, duration)); } #[cfg(feature = "cloud")] - apply_completed_job_cloud_usage(db, queued_job, _duration); + apply_completed_job_cloud_usage(db, queued_job, duration); #[cfg(feature = "enterprise")] apply_completed_job_error_handlers( @@ -820,7 +820,7 @@ pub async fn add_completed_job( // tracing::error!("4 {:?}", start.elapsed()); - Ok(queued_job.id) + Ok((queued_job.id, duration)) } async fn commit_completed_job( @@ -4982,12 +4982,17 @@ async fn restarted_flows_resolution( if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() { new_flow_jobs_success.truncate(branch_or_iteration_n); } + let mut new_flow_jobs_timeline = module.flow_jobs_duration(); + if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() { + new_flow_jobs_timeline.truncate(branch_or_iteration_n); + } truncated_modules.push(FlowStatusModule::InProgress { id: module.id(), job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow iterator: None, flow_jobs: Some(new_flow_jobs), flow_jobs_success: new_flow_jobs_success, + flow_jobs_duration: new_flow_jobs_timeline, branch_chosen: None, branchall: Some(BranchAllStatus { branch: branch_or_iteration_n - 1, // Doing minus one here as this variable reflects the latest finished job in the iteration @@ -5022,6 +5027,10 @@ async fn restarted_flows_resolution( if let Some(new_flow_jobs_success) = new_flow_jobs_success.as_mut() { new_flow_jobs_success.truncate(branch_or_iteration_n); } + let mut new_flow_jobs_timeline = module.flow_jobs_duration(); + if let Some(new_flow_jobs_timeline) = new_flow_jobs_timeline.as_mut() { + new_flow_jobs_timeline.truncate(branch_or_iteration_n); + } truncated_modules.push(FlowStatusModule::InProgress { id: module.id(), job: new_flow_jobs[new_flow_jobs.len() - 1], // set to last finished job from completed flow @@ -5031,6 +5040,7 @@ async fn restarted_flows_resolution( }), flow_jobs: Some(new_flow_jobs), flow_jobs_success: new_flow_jobs_success, + flow_jobs_duration: new_flow_jobs_timeline, branch_chosen: None, branchall: None, parallel, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 591d6bf11f..144c8d9101 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -17,13 +17,7 @@ use windmill_common::otel_oss::FutureExt; use uuid::Uuid; use windmill_common::{ - add_time, - error::{self, Error}, - jobs::JobKind, - utils::WarnAfterExt, - worker::{to_raw_value, Connection, WORKER_GROUP}, - worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, - KillpillSender, DB, + add_time, error::{self, Error}, flow_status::{FlowJobDuration}, jobs::JobKind, utils::WarnAfterExt, worker::{to_raw_value, Connection, WORKER_GROUP}, worker_group_job_stats::{accumulate_job_stats, flush_stats_to_db, JobStatsMap}, KillpillSender, DB }; #[cfg(feature = "benchmark")] @@ -342,6 +336,7 @@ pub fn start_background_processor( &w_id, success, Arc::new(result), + None, true, &same_worker_tx, &worker_dir, @@ -582,6 +577,7 @@ pub async fn process_completed_job( let parent_job = job.parent_job.clone(); let job_id = job.id.clone(); let workspace_id = job.workspace_id.clone(); + let started_at = job.started_at.clone(); if job.flow_step_id.as_deref() == Some("preprocessor") { // Do this before inserting to `v2_job_completed` for backwards compatibility @@ -614,7 +610,7 @@ pub async fn process_completed_job( add_time!(bench, "pre add_completed_job"); - add_completed_job( + let (_, duration) = add_completed_job( db, &job, true, @@ -642,6 +638,7 @@ pub async fn process_completed_job( &workspace_id, true, result, + started_at.map(|x| FlowJobDuration { started_at: x, duration_ms: duration }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, @@ -682,6 +679,7 @@ pub async fn process_completed_job( &job.workspace_id, false, Arc::new(serde_json::value::to_raw_value(&result).unwrap()), + duration.map(|x| FlowJobDuration { started_at: job.started_at.unwrap(), duration_ms: x }), false, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).to_owned(), &worker_dir, @@ -783,6 +781,7 @@ pub async fn handle_job_error( &job.workspace_id, false, Arc::new(serde_json::value::to_raw_value(&wrapped_error).unwrap()), + None, unrecoverable, &same_worker_tx.expect(SAME_WORKER_REQUIREMENTS).clone(), worker_dir, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 201924e825..832af368d1 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -37,7 +37,8 @@ use windmill_common::cache::{self, RawData}; use windmill_common::client::AuthedClient; use windmill_common::db::Authed; use windmill_common::flow_status::{ - ApprovalConditions, FlowStatusModuleWParent, Iterator as FlowIterator, JobResult, + ApprovalConditions, FlowJobDuration, FlowJobsDuration, FlowStatusModuleWParent, + Iterator as FlowIterator, JobResult, }; use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf}; use windmill_common::jobs::{ @@ -79,6 +80,7 @@ pub async fn update_flow_status_after_job_completion( w_id: &str, success: bool, result: Arc>, + flow_job_duration: Option, unrecoverable: bool, same_worker_tx: &SameWorkerSender, worker_dir: &str, @@ -95,6 +97,7 @@ pub async fn update_flow_status_after_job_completion( job_id_for_status: job_id_for_status.clone(), success, result, + flow_job_duration, stop_early_override, has_triggered_error_handler: false, }; @@ -108,6 +111,7 @@ pub async fn update_flow_status_after_job_completion( &rec.job_id_for_status, w_id, rec.success, + rec.flow_job_duration.clone(), rec.result, unrecoverable, same_worker_tx, @@ -131,6 +135,7 @@ pub async fn update_flow_status_after_job_completion( &rec.job_id_for_status, w_id, false, + rec.flow_job_duration, Arc::new(to_raw_value(&Json(&WrappedError { error: json!(e.to_string()), }))), @@ -180,11 +185,13 @@ pub enum UpdateFlowStatusAfterJobCompletion { NonLastParallelBranch, PreprocessingStep, } + pub struct RecUpdateFlowStatusAfterJobCompletion { flow: uuid::Uuid, job_id_for_status: Uuid, success: bool, result: Arc>, + flow_job_duration: Option, stop_early_override: Option, has_triggered_error_handler: bool, } @@ -267,6 +274,7 @@ pub async fn update_flow_status_after_job_completion_internal( job_id_for_status: &Uuid, w_id: &str, mut success: bool, + mut flow_job_duration: Option, result: Arc>, unrecoverable: bool, same_worker_tx: &SameWorkerSender, @@ -508,6 +516,7 @@ pub async fn update_flow_status_after_job_completion_internal( parallel, flow_jobs: Some(jobs), flow_jobs_success, + flow_jobs_duration, .. } if *parallel => { let (nindex, len) = match (iterator, branchall) { @@ -517,21 +526,38 @@ pub async fn update_flow_status_after_job_completion_internal( } else { None }; - + tracing::error!( + "flow_job_duration: {:?}, position: {:?} flow: {:?}", + flow_job_duration, + position, + flow + ); let nindex = if let Some(position) = position { sqlx::query_scalar!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + "UPDATE v2_job_status SET flow_status = + CASE + WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET( + JSONB_SET(JSONB_SET(JSONB_SET( + flow_status, + ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6), + ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb + ) + ELSE + JSONB_SET(JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), ARRAY['modules', $1::TEXT, 'iterator', 'index'], ((flow_status->'modules'->$1::int->'iterator'->>'index')::int + 1)::text::jsonb - ) + ) END WHERE id = $2 RETURNING (flow_status->'modules'->$1::int->'iterator'->>'index')::int", old_status.step, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.started_at)), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) ) } else { sqlx::query_scalar!( @@ -554,6 +580,19 @@ pub async fn update_flow_status_after_job_completion_internal( )) })? .ok_or_else(|| Error::internal_err(format!("requiring an index in InProgress")))?; + + // let status_for_debug = sqlx::query!( + // "SELECT flow_status FROM v2_job_status WHERE id = $1", + // flow + // ) + // .fetch_one(&mut *tx) + // .await + // .map_err(|e| { + // Error::internal_err(format!("error while fetching flow status: {e:#}")) + // })?; + + // tracing::error!("status_for_debug: {:?}", status_for_debug.flow_status); + tracing::info!( "parallel iteration {job_id_for_status} of flow {flow} update nindex: {nindex} len: {len}", nindex = nindex, @@ -570,18 +609,33 @@ pub async fn update_flow_status_after_job_completion_internal( let nindex = if let Some(position) = position { sqlx::query_scalar!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( - JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + "UPDATE v2_job_status SET flow_status = + CASE + WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET( + JSONB_SET(JSONB_SET(JSONB_SET( + flow_status, + ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $6), + ARRAY['modules', $1::TEXT, 'branchall', 'branch'], + ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb + ) + ELSE + JSONB_SET(JSONB_SET( + flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4), ARRAY['modules', $1::TEXT, 'branchall', 'branch'], ((flow_status->'modules'->$1::int->'branchall'->>'branch')::int + 1)::text::jsonb ) + END WHERE id = $2 RETURNING (flow_status->'modules'->$1::int->'branchall'->>'branch')::int", old_status.step, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.started_at)), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)) ) } else { sqlx::query_scalar!( @@ -616,7 +670,22 @@ pub async fn update_flow_status_after_job_completion_internal( .execute(&mut *tx) .await?; if nindex == len { - let mut flow_jobs_success = flow_jobs_success.clone(); + let success_and_durations = match sqlx::query!( + "SELECT flow_status->'modules'->$2::int->'flow_jobs_success' as \"flow_jobs_success: Json>>\", flow_status->'modules'->$2::int->'flow_jobs_duration' as \"flow_jobs_duration: Json\" + FROM v2_job_status WHERE id = $1", + flow, + old_status.step + ) + .fetch_one(&mut *tx) + .await { + Err(e) => { + tracing::error!("error while fetching success and durations: {e:#}"); + (flow_jobs_success.clone(), flow_jobs_duration.clone()) + } + Ok(x) => (x.flow_jobs_success.map(|x| x.0), x.flow_jobs_duration.map(|x| x.0)), + }; + + let mut flow_jobs_success = success_and_durations.0; if let Some(flow_job_success) = flow_jobs_success.as_mut() { let position = jobs.iter().position(|x| x == job_id_for_status); if let Some(position) = position { @@ -625,6 +694,11 @@ pub async fn update_flow_status_after_job_completion_internal( } } } + let mut flow_jobs_duration = success_and_durations.1; + if let Some(flow_jobs_duration) = flow_jobs_duration.as_mut() { + let position = jobs.iter().position(|x| x == job_id_for_status); + flow_jobs_duration.set(position, &flow_job_duration); + } let branches = current_module .and_then(|x| x.get_branches_skip_failures().ok()) @@ -698,6 +772,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs: Some(jobs.clone()), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), branch_chosen: None, approvers: vec![], failed_retries: vec![], @@ -712,6 +787,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs: Some(jobs.clone()), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), branch_chosen: None, failed_retries: vec![], agent_actions: None, @@ -797,13 +873,14 @@ pub async fn update_flow_status_after_job_completion_internal( && !stop_early => { if let Some(jobs) = flow_jobs { - set_success_in_flow_job_success( + set_success_and_duration_in_flow_job_success( flow_jobs_success, jobs, job_id_for_status, - &old_status, + old_status.step, flow, success, + flow_job_duration.clone(), &mut tx, ) .await?; @@ -821,13 +898,14 @@ pub async fn update_flow_status_after_job_completion_internal( && !stop_early => { if let Some(jobs) = flow_jobs { - set_success_in_flow_job_success( + set_success_and_duration_in_flow_job_success( flow_jobs_success, jobs, job_id_for_status, - &old_status, + old_status.step, flow, success, + flow_job_duration.clone(), &mut tx, ) .await?; @@ -872,6 +950,7 @@ pub async fn update_flow_status_after_job_completion_internal( let flow_jobs = module_status.flow_jobs(); let branch_chosen = module_status.branch_chosen(); let mut flow_jobs_success = module_status.flow_jobs_success(); + let mut flow_jobs_duration = module_status.flow_jobs_duration(); if let (Some(flow_job_success), Some(flow_jobs)) = (flow_jobs_success.as_mut(), flow_jobs.as_ref()) @@ -884,6 +963,13 @@ pub async fn update_flow_status_after_job_completion_internal( } } + if let (Some(flow_jobs_duration), Some(flow_jobs)) = + (flow_jobs_duration.as_mut(), flow_jobs.as_ref()) + { + let position = flow_jobs.iter().position(|x| x == job_id_for_status); + flow_jobs_duration.set(position, &flow_job_duration); + } + // if stop_early with error message, we want to set the job as failure and trigger the error handler if it exists if (success || (flow_jobs.is_some() && (skip_loop_failures || skip_seq_branch_failure))) @@ -912,6 +998,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs, flow_jobs_success, + flow_jobs_duration, branch_chosen, approvers: vec![], failed_retries: old_status.retry.failed_jobs.clone(), @@ -953,6 +1040,7 @@ pub async fn update_flow_status_after_job_completion_internal( job: job_id_for_status.clone(), flow_jobs, flow_jobs_success, + flow_jobs_duration, branch_chosen, failed_retries: old_status.retry.failed_jobs.clone(), agent_actions: module_status.agent_actions(), @@ -1387,8 +1475,8 @@ pub async fn update_flow_status_after_job_completion_internal( let success = success && (!is_failure_step || result_has_recover_true(nresult.clone())); add_time!(bench, "flow status update 1"); - if success { - add_completed_job( + let duration = if success { + let (_, duration) = add_completed_job( db, &flow_job, true, @@ -1401,8 +1489,9 @@ pub async fn update_flow_status_after_job_completion_internal( None, ) .await?; + duration } else { - add_completed_job( + let (_, duration) = add_completed_job( db, &flow_job, false, @@ -1419,7 +1508,11 @@ pub async fn update_flow_status_after_job_completion_internal( None, ) .await?; - } + duration + }; + flow_job_duration = flow_job + .started_at + .map(|x| FlowJobDuration { started_at: x, duration_ms: duration }); } true } else { @@ -1469,6 +1562,7 @@ pub async fn update_flow_status_after_job_completion_internal( flow: parent_job, job_id_for_status: flow, success: success && !is_failure_step, + flow_job_duration: flow_job_duration.clone(), result: nresult.clone(), stop_early_override: if stop_early { Some(skip_if_stop_early) @@ -1490,35 +1584,46 @@ fn find_flow_job_index(flow_jobs: &Vec, job_id_for_status: &Uuid) -> Optio flow_jobs.iter().position(|x| x == job_id_for_status) } -async fn set_success_in_flow_job_success<'c>( +async fn set_success_and_duration_in_flow_job_success<'c>( flow_jobs_success: &Option>>, flow_jobs: &Vec, job_id_for_status: &Uuid, - old_status: &FlowStatus, + old_status_step: i32, flow: Uuid, success: bool, + flow_job_duration: Option, tx: &mut Transaction<'c, Postgres>, ) -> error::Result<()> { if flow_jobs_success.is_some() { let position = find_flow_job_index(flow_jobs, job_id_for_status); if let Some(position) = position { sqlx::query!( - "UPDATE v2_job_status SET - flow_status = JSONB_SET( + "UPDATE v2_job_status SET flow_status = + CASE WHEN flow_status->'modules'->$1::TEXT->'flow_jobs_duration' IS NOT NULL THEN + JSONB_SET(JSONB_SET(JSONB_SET( flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4 - ) - WHERE id = $2", - old_status.step as i32, + ), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'duration_ms', $3::TEXT], $5), + ARRAY['modules', $1::TEXT, 'flow_jobs_duration', 'started_at', $3::TEXT], $6) + ELSE + JSONB_SET(flow_status, ARRAY['modules', $1::TEXT, 'flow_jobs_success', $3::TEXT], $4) + END + WHERE id = $2", + old_status_step as i32, flow, position as i32, - json!(success) + json!(success), + flow_job_duration.as_ref().map(|x| json!(x.duration_ms)), + flow_job_duration.as_ref().map(|x| json!(x.started_at)) ) .execute(&mut **tx) .await .map_err(|e| { - Error::internal_err(format!("error while setting flow_jobs_success: {e:#}")) + Error::internal_err(format!( + "error while setting flow_jobs_success/timeline: {e:#}" + )) })?; } } @@ -2753,6 +2858,11 @@ async fn push_next_flow_job( } else { Some(vec![]) }, + flow_jobs_duration: if branch_chosen.is_some() { + None + } else { + Some(FlowJobsDuration { started_at: vec![], duration_ms: vec![] }) + }, branch_chosen: branch_chosen, approvers: vec![], failed_retries: vec![], @@ -3148,6 +3258,7 @@ async fn push_next_flow_job( mut flow_jobs, while_loop, mut flow_jobs_success, + mut flow_jobs_duration, .. }, .. @@ -3159,11 +3270,15 @@ async fn push_next_flow_job( if let Some(flow_jobs_success) = &mut flow_jobs_success { flow_jobs_success.push(None); } + if let Some(flow_jobs_duration) = &mut flow_jobs_duration { + flow_jobs_duration.push(&None); + } FlowStatusModule::InProgress { job: uuid, iterator: Some(FlowIterator { index, itered }), flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, branch_chosen: None, branchall: None, id: status_module.id(), @@ -3179,6 +3294,7 @@ async fn push_next_flow_job( iterator, flow_jobs_success: Some(vec![None; uuids.len()]), flow_jobs: Some(uuids.clone()), + flow_jobs_duration: Some(FlowJobsDuration::new(uuids.len())), branch_chosen: None, branchall, id: status_module.id(), @@ -3192,6 +3308,7 @@ async fn push_next_flow_job( mut flow_jobs, status, mut flow_jobs_success, + mut flow_jobs_duration, .. }) => { let uuid = one_uuid?; @@ -3199,11 +3316,15 @@ async fn push_next_flow_job( if let Some(flow_jobs_success) = &mut flow_jobs_success { flow_jobs_success.push(None); } + if let Some(flow_jobs_duration) = &mut flow_jobs_duration { + flow_jobs_duration.push(&None); + } FlowStatusModule::InProgress { job: uuid, iterator: None, flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, branch_chosen: None, branchall: Some(status), id: status_module.id(), @@ -3220,6 +3341,7 @@ async fn push_next_flow_job( iterator: None, flow_jobs: None, flow_jobs_success: None, + flow_jobs_duration: None, branch_chosen: Some(branch), branchall: None, id: status_module.id(), @@ -3396,6 +3518,7 @@ struct ForloopNextIteration { itered: Vec>, flow_jobs: Vec, flow_jobs_success: Option>>, + flow_jobs_duration: Option, new_args: Iter, while_loop: bool, } @@ -3411,6 +3534,7 @@ struct NextBranch { status: BranchAllStatus, flow_jobs: Vec, flow_jobs_success: Option>>, + flow_jobs_duration: Option, } #[derive(Debug)] @@ -3662,13 +3786,18 @@ async fn compute_next_flow_transform( FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => { // if it's a simple single step flow, we will collapse it as an optimization and need to pass flow_input as an arg let is_simple = is_simple_modules(&modules, flow.failure_module.as_ref()); - let (flow_jobs, flow_jobs_success) = match status_module { + let (flow_jobs, flow_jobs_success, flow_jobs_duration) = match status_module { FlowStatusModule::InProgress { flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, .. - } => (flow_jobs.clone(), flow_jobs_success.clone()), - _ => (vec![], Some(vec![])), + } => ( + flow_jobs.clone(), + flow_jobs_success.clone(), + flow_jobs_duration.clone(), + ), + _ => (vec![], Some(vec![]), Some(FlowJobsDuration::new(0))), }; let next_loop_idx = flow_jobs.len(); next_loop_iteration( @@ -3679,6 +3808,7 @@ async fn compute_next_flow_transform( itered: vec![], flow_jobs: flow_jobs, flow_jobs_success: flow_jobs_success, + flow_jobs_duration: flow_jobs_duration, new_args: Iter { index: next_loop_idx as i32, value: windmill_common::worker::to_raw_value(&next_loop_idx), @@ -3877,72 +4007,78 @@ async fn compute_next_flow_transform( )) } FlowModuleValue::BranchAll { branches, parallel, .. } => { - let (branch_status, flow_jobs, flow_jobs_success) = match status_module { - FlowStatusModule::WaitingForPriorSteps { .. } - | FlowStatusModule::WaitingForEvents { .. } - | FlowStatusModule::WaitingForExecutor { .. } => { - if branches.is_empty() { - return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None }); - } else if parallel { - let len = branches.len(); - let payloads: Vec = branches - .into_iter() - .enumerate() - .filter_map(|(i, Branch { modules, modules_node, .. })| { - let Some(payload) = payload_from_modules( - modules, - modules_node, - flow.failure_module.as_ref(), - flow.same_worker, - || format!("{}-{i}", status.step), - || format!("{}/branchall-{}", flow_job.runnable_path(), i), - false, - ) else { - return None; - }; - Some(JobPayloadWithTag { - payload, - tag: None, - delete_after_use, - timeout: None, - on_behalf_of: None, - }) - }) - .collect::>(); - if payloads.is_empty() { + let (branch_status, flow_jobs, flow_jobs_success, flow_jobs_duration) = + match status_module { + FlowStatusModule::WaitingForPriorSteps { .. } + | FlowStatusModule::WaitingForEvents { .. } + | FlowStatusModule::WaitingForExecutor { .. } => { + if branches.is_empty() { return Ok(NextFlowTransform::EmptyInnerFlows { branch_chosen: None }); + } else if parallel { + let len = branches.len(); + let payloads: Vec = branches + .into_iter() + .enumerate() + .filter_map(|(i, Branch { modules, modules_node, .. })| { + let Some(payload) = payload_from_modules( + modules, + modules_node, + flow.failure_module.as_ref(), + flow.same_worker, + || format!("{}-{i}", status.step), + || format!("{}/branchall-{}", flow_job.runnable_path(), i), + false, + ) else { + return None; + }; + Some(JobPayloadWithTag { + payload, + tag: None, + delete_after_use, + timeout: None, + on_behalf_of: None, + }) + }) + .collect::>(); + if payloads.is_empty() { + return Ok(NextFlowTransform::EmptyInnerFlows { + branch_chosen: None, + }); + } + return Ok(NextFlowTransform::Continue( + ContinuePayload::ParallelJobs(payloads), + NextStatus::AllFlowJobs { + branchall: Some(BranchAllStatus { branch: 0, len }), + iterator: None, + simple_input_transforms: None, + }, + )); + } else { + ( + BranchAllStatus { branch: 0, len: branches.len() }, + vec![], + Some(vec![]), + Some(FlowJobsDuration::new(0)), + ) } - return Ok(NextFlowTransform::Continue( - ContinuePayload::ParallelJobs(payloads), - NextStatus::AllFlowJobs { - branchall: Some(BranchAllStatus { branch: 0, len }), - iterator: None, - simple_input_transforms: None, - }, - )); - } else { - ( - BranchAllStatus { branch: 0, len: branches.len() }, - vec![], - Some(vec![]), - ) } - } - FlowStatusModule::InProgress { - branchall: Some(BranchAllStatus { branch, len }), - flow_jobs: Some(flow_jobs), - flow_jobs_success, - .. - } if !parallel => ( - BranchAllStatus { branch: branch + 1, len: len.clone() }, - flow_jobs.clone(), - flow_jobs_success.clone(), - ), + FlowStatusModule::InProgress { + branchall: Some(BranchAllStatus { branch, len }), + flow_jobs: Some(flow_jobs), + flow_jobs_success, + flow_jobs_duration, + .. + } if !parallel => ( + BranchAllStatus { branch: branch + 1, len: len.clone() }, + flow_jobs.clone(), + flow_jobs_success.clone(), + flow_jobs_duration.clone(), + ), - _ => Err(Error::BadRequest(format!( - "Unrecognized module status for BranchAll {status_module:?}" - )))?, - }; + _ => Err(Error::BadRequest(format!( + "Unrecognized module status for BranchAll {status_module:?}" + )))?, + }; let Branch { modules, modules_node, .. } = branches .into_iter() @@ -3972,7 +4108,6 @@ async fn compute_next_flow_transform( branch_chosen: Some(BranchChosen::Default), }); }; - Ok(NextFlowTransform::Continue( ContinuePayload::SingleJob(JobPayloadWithTag { payload, @@ -3985,6 +4120,7 @@ async fn compute_next_flow_transform( status: branch_status, flow_jobs, flow_jobs_success, + flow_jobs_duration, }), )) } @@ -4130,6 +4266,7 @@ async fn next_forloop_status( itered, flow_jobs: vec![], flow_jobs_success: Some(vec![]), + flow_jobs_duration: Some(FlowJobsDuration::new(0)), new_args: iter, while_loop: false, }) @@ -4142,6 +4279,7 @@ async fn next_forloop_status( iterator: Some(FlowIterator { itered, index }), flow_jobs: Some(flow_jobs), flow_jobs_success, + flow_jobs_duration, .. } if !*parallel => { let itered_new = if itered.is_empty() { @@ -4192,6 +4330,7 @@ async fn next_forloop_status( itered: itered_new.clone(), flow_jobs: flow_jobs.clone(), flow_jobs_success: flow_jobs_success.clone(), + flow_jobs_duration: flow_jobs_duration.clone(), new_args: Iter { index: index as i32, value: next.to_owned() }, while_loop: false, }) diff --git a/cli/src/commands/sync/sync.ts b/cli/src/commands/sync/sync.ts index 4ab1af2252..bb641b5e0d 100644 --- a/cli/src/commands/sync/sync.ts +++ b/cli/src/commands/sync/sync.ts @@ -65,7 +65,10 @@ import { } from "../../utils/metadata.ts"; import { OpenFlow } from "../../../gen/types.gen.ts"; import { pushResource } from "../resource/resource.ts"; -import { newPathAssigner, PathAssigner } from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts"; +import { + newPathAssigner, + PathAssigner, +} from "../../../windmill-utils-internal/src/path-utils/path-assigner.ts"; import { extractInlineScripts as extractInlineScriptsForFlows } from "../../../windmill-utils-internal/src/inline-scripts/extractor.ts"; // Merge CLI options with effective settings, preserving CLI flags as overrides @@ -158,7 +161,9 @@ async function addCodebaseDigestIfRelevant( try { parsed = yamlParseContent(path, content); } catch (error) { - log.error(`Failed to parse YAML content for codebase digest at path: ${path}`); + log.error( + `Failed to parse YAML content for codebase digest at path: ${path}` + ); throw error; } if (parsed && typeof parsed == "object") { @@ -258,7 +263,10 @@ export interface InlineScript { content: string; } -export function extractInlineScriptsForApps(rec: any, pathAssigner: PathAssigner): InlineScript[] { +export function extractInlineScriptsForApps( + rec: any, + pathAssigner: PathAssigner +): InlineScript[] { if (!rec) { return []; } @@ -349,10 +357,12 @@ function ZipFSElement( flow.value.modules, {}, SEP, - defaultTs, + defaultTs ); } catch (error) { - log.error(`Failed to extract inline scripts for flow at path: ${p}`); + log.error( + `Failed to extract inline scripts for flow at path: ${p}` + ); throw error; } for (const s of inlineScripts) { @@ -386,9 +396,14 @@ function ZipFSElement( } let inlineScripts; try { - inlineScripts = extractInlineScriptsForApps(app?.["value"], newPathAssigner(defaultTs)); + inlineScripts = extractInlineScriptsForApps( + app?.["value"], + newPathAssigner(defaultTs) + ); } catch (error) { - log.error(`Failed to extract inline scripts for app at path: ${p}`); + log.error( + `Failed to extract inline scripts for app at path: ${p}` + ); throw error; } for (const s of inlineScripts) { @@ -913,13 +928,17 @@ async function compareDynFSElement( try { parsedV = JSON.parse(v); } catch (error) { - log.error(`Failed to parse new JSON content for comparison at path: ${k}`); + log.error( + `Failed to parse new JSON content for comparison at path: ${k}` + ); throw error; } try { parsedM2 = JSON.parse(m2[k]); } catch (error) { - log.error(`Failed to parse existing JSON content for comparison at path: ${k}`); + log.error( + `Failed to parse existing JSON content for comparison at path: ${k}` + ); throw error; } if (deepEqual(parsedV, parsedM2)) { @@ -932,11 +951,11 @@ async function compareDynFSElement( continue; } if (!ignoreCodebaseChanges) { - if (before.codebase != undefined) { + if (before?.codebase != undefined) { delete before.codebase; m2[k] = yamlStringify(before, yamlOptions); } - if (after.codebase != undefined) { + if (after?.codebase != undefined) { if (before.codebase != after.codebase) { codebaseChanges[k] = after.codebase; } @@ -1241,7 +1260,6 @@ export async function pull( opts: GlobalOptions & SyncOptions & { repository?: string; promotion?: string } ) { - const originalCliOpts = { ...opts }; opts = await mergeConfigWithConfigFile(opts); @@ -1345,7 +1363,10 @@ export async function pull( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1380,7 +1401,10 @@ export async function pull( // Determine if this file should be written to a branch-specific path let targetPath = change.path; if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { targetPath = branchSpecificPath; } @@ -1430,12 +1454,24 @@ export async function pull( } } if (exts.some((e) => change.path.endsWith(e))) { - log.info(`Editing script content of ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Editing script content of ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } else if ( change.path.endsWith(".yaml") || change.path.endsWith(".json") ) { - log.info(`Editing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Editing ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } await Deno.writeTextFile(target, change.after); @@ -1447,10 +1483,22 @@ export async function pull( await ensureDir(path.dirname(target)); if (opts.stateful) { await ensureDir(path.dirname(stateTarget)); - log.info(`Adding ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Adding ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); } await Deno.writeTextFile(target, change.content); - log.info(`Writing ${getTypeStrFromPath(change.path)} ${targetPath}${targetPath !== change.path ? colors.gray(` (branch-specific override for ${change.path})`) : ""}`); + log.info( + `Writing ${getTypeStrFromPath(change.path)} ${targetPath}${ + targetPath !== change.path + ? colors.gray(` (branch-specific override for ${change.path})`) + : "" + }` + ); if (opts.stateful) { await Deno.copyFile(target, stateTarget); } @@ -1528,7 +1576,10 @@ export async function pull( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1560,7 +1611,10 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) { // Check if this will be written as a branch-specific file if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { displayPath = branchSpecificPath; branchNote = " (branch-specific)"; @@ -1569,17 +1623,26 @@ function prettyChanges(changes: Change[], specificItems?: SpecificItemsConfig) { if (change.name === "added") { log.info( - colors.green(`+ ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote)) + colors.green( + `+ ${getTypeStrFromPath(change.path)} ` + + displayPath + + colors.gray(branchNote) + ) ); } else if (change.name === "deleted") { log.info( - colors.red(`- ${getTypeStrFromPath(change.path)} ` + displayPath + colors.gray(branchNote)) + colors.red( + `- ${getTypeStrFromPath(change.path)} ` + + displayPath + + colors.gray(branchNote) + ) ); } else if (change.name === "edited") { log.info( colors.yellow( `~ ${getTypeStrFromPath(change.path)} ` + - displayPath + colors.gray(branchNote) + + displayPath + + colors.gray(branchNote) + (change.codebase ? ` (codebase changed)` : "") ) ); @@ -1794,7 +1857,10 @@ export async function push( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), @@ -1938,7 +2004,10 @@ export async function push( const currentBranch = getCurrentGitBranch(); if (currentBranch && isBranchSpecificFile(resourceFilePath)) { - serverPath = fromBranchSpecificPath(resourceFilePath, currentBranch); + serverPath = fromBranchSpecificPath( + resourceFilePath, + currentBranch + ); } await pushResource( @@ -1960,7 +2029,10 @@ export async function push( // Check if this is a branch-specific item and get the original branch-specific path let originalBranchSpecificPath: string | undefined; if (specificItems && isSpecificItem(change.path, specificItems)) { - originalBranchSpecificPath = getBranchSpecificPath(change.path, specificItems); + originalBranchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); } await pushObj( @@ -2010,7 +2082,10 @@ export async function push( // For branch-specific items, we read from branch-specific files but push to base server paths let localFilePath = change.path; if (specificItems && isSpecificItem(change.path, specificItems)) { - const branchSpecificPath = getBranchSpecificPath(change.path, specificItems); + const branchSpecificPath = getBranchSpecificPath( + change.path, + specificItems + ); if (branchSpecificPath) { localFilePath = branchSpecificPath; } @@ -2024,7 +2099,7 @@ export async function push( opts.plainSecrets ?? false, [], opts.message, - localFilePath // Pass the actual local file path + localFilePath // Pass the actual local file path ); if (stateTarget) { @@ -2216,7 +2291,10 @@ export async function push( ...(specificItems && isSpecificItem(change.path, specificItems) ? { branch_specific: true, - branch_specific_path: getBranchSpecificPath(change.path, specificItems) + branch_specific_path: getBranchSpecificPath( + change.path, + specificItems + ), } : {}), })), diff --git a/frontend/openapi-ts-error-1758271586180.log b/frontend/openapi-ts-error-1758271586180.log new file mode 100644 index 0000000000..4c0634a7c7 --- /dev/null +++ b/frontend/openapi-ts-error-1758271586180.log @@ -0,0 +1,28 @@ +Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25) + + 8022 | description: job args + 8023 | content: + 8024 | application/json: + 8025 | schema: {}\ +--------------------------------^ + 8026 | + 8027 | /w/{workspace}/jobs/queue/get_scheduled_for_by_ids: +ParserError: Error parsing /home/rfiszel/windmill/backend/windmill-api/openapi.yaml: bad indentation of a mapping entry (8025:25) + + 8022 | description: job args + 8023 | content: + 8024 | application/json: + 8025 | schema: {}\ +--------------------------------^ + 8026 | + 8027 | /w/{workspace}/jobs/queue/get_scheduled_for_by_ids: + at Object.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parsers/yaml.js:44:23) + at getResult (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:116:22) + at runNextPlugin (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:64:32) + at /home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:55:9 + at new Promise () + at Object.run (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/util/plugins.js:54:12) + at parseFile (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:130:38) + at parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/parse.js:56:30) + at async $RefParser.parse (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:115:28) + at async $RefParser.resolve (/home/rfiszel/windmill/frontend/node_modules/@apidevtools/json-schema-ref-parser/dist/lib/index.js:145:13) \ No newline at end of file diff --git a/frontend/src/lib/components/FlowLogViewer.svelte b/frontend/src/lib/components/FlowLogViewer.svelte index a22e0a4211..6374f69930 100644 --- a/frontend/src/lib/components/FlowLogViewer.svelte +++ b/frontend/src/lib/components/FlowLogViewer.svelte @@ -19,7 +19,7 @@ import { twMerge } from 'tailwind-merge' import FlowJobsMenu from './flows/map/FlowJobsMenu.svelte' import BarsStaggered from './icons/BarsStaggered.svelte' - import type { GlobalIterationBounds, GraphModuleState } from './graph/model' + import type { GraphModuleState } from './graph/model' import type { NavigationChain } from '$lib/keyboardChain' import { updateLinks } from '$lib/keyboardChain' import FlowLogRow from './FlowLogRow.svelte' @@ -62,8 +62,6 @@ timelineAvailableWidths: Record timelinelWidth: number showTimeline?: boolean - globalIterationBounds?: Record - loadPreviousIterations?: (key: string, amount: number) => void } let { @@ -92,9 +90,7 @@ timelineNow, timelineAvailableWidths = $bindable(), timelinelWidth, - showTimeline = true, - globalIterationBounds, - loadPreviousIterations + showTimeline = true }: Props = $props() function getJobLink(jobId: string | undefined): string { @@ -769,13 +765,8 @@ total={timelineTotal} min={timelineMin} items={moduleItems ?? []} - hasMoreIterations={globalIterationBounds?.[module.id] && - (globalIterationBounds[module.id].iteration_from ?? 0) > 0} now={timelineNow} {timelinelWidth} - loadPreviousIterations={() => { - loadPreviousIterations?.(module.id, 20) - }} onSelectIteration={(id) => { if ( module.value.type !== 'forloopflow' && @@ -856,8 +847,6 @@ bind:timelineAvailableWidths {timelinelWidth} {showTimeline} - {globalIterationBounds} - {loadPreviousIterations} /> {/each} diff --git a/frontend/src/lib/components/FlowLogViewerWrapper.svelte b/frontend/src/lib/components/FlowLogViewerWrapper.svelte index 274290ec1f..4549fc8304 100644 --- a/frontend/src/lib/components/FlowLogViewerWrapper.svelte +++ b/frontend/src/lib/components/FlowLogViewerWrapper.svelte @@ -1,6 +1,6 @@ { - timelineCompute?.updateInputs(flowModules, durationStatuses, flowDone) + timelineCompute?.updateInputs(flowModulesIds, durationStatuses, flowDone) }} /> - {#if items}
{/if}
{max ? displayDate(new Date(max), true) : ''}{#if !max && min}{#if now} - {msToSec(now - min, 3)}s + {msToSec(now - min, 1)}s {/if}{/if}
@@ -109,61 +118,87 @@ />
{/if} - {#each Object.values(flowModules) as k (k)} - {@const iterationFrom = globalIterationBounds[buildSubflowKey(k)]?.iteration_from ?? 0} -
- {#if iterationFrom > 0} -
- -
- {/if} - + {#each flowModules as { id: k, type: typ } (k)} + {@const subItems = items?.[k]?.filter((x) => x.created_at && x.started_at)} +
-
{k.startsWith('subflow:') ? k.substring(8) : k}
-
+
{k.startsWith('subflow:') ? k.substring(8) : k} + {#if localModuleStates[k]?.selectedForloop && (typ == 'forloopflow' || typ == 'whileloopflow')} + + + + {/if} +
+
+ {#if subItems?.length > 1} +
+ {subItems?.length} jobs +
+ {/if} {#if min && total} -
- {#each items?.[k] ?? [] as b} - {@const waitingLen = b?.created_at - ? b.started_at - ? b.started_at - b?.created_at - : b.duration_ms - ? 0 - : now - b?.created_at - : 0} -
- - {#if b.started_at} + subItems?.[index]?.id} + > + {#snippet item({ index, style })} + {@const b = subItems?.[index]} + {#if b?.created_at} + + {@const waitingLen = b?.created_at + ? b.started_at + ? b.started_at - b?.created_at + : b.duration_ms + ? 0 + : now - b?.created_at + : 0} +
- {/if} -
- {/each} -
+ {#if b.started_at} + + {/if} +
+ {:else} +
+
+ +
+
+ {/if} + {/snippet} + {/if}
diff --git a/frontend/src/lib/components/FlowTimelineBar.svelte b/frontend/src/lib/components/FlowTimelineBar.svelte index a78a9fa18a..12ad614049 100644 --- a/frontend/src/lib/components/FlowTimelineBar.svelte +++ b/frontend/src/lib/components/FlowTimelineBar.svelte @@ -21,8 +21,6 @@ showZoomButtons?: boolean onZoom?: () => void zoom?: 'in' | 'out' - hasMoreIterations?: boolean - loadPreviousIterations?: () => void onSelectIteration?: (id: string) => void idToIterationIndex?: (id: string) => number | undefined showIterations?: string[] @@ -39,8 +37,6 @@ showZoomButtons = false, onZoom, zoom = 'in', - hasMoreIterations, - loadPreviousIterations, onSelectIteration, idToIterationIndex, showIterations, @@ -112,26 +108,61 @@ return { left: leftPercent, width: widthPercent } } - function getOverlapOpacity(item: TimelineItem, allItems: TimelineItem[]): number { - if (!item.started_at) return 1 + // More efficient version using sweep line algorithm for computing all overlaps at once + function computeAllOverlaps(items: TimelineItem[]): Record { + const overlapCounts = new Map() - const itemEnd = item.duration_ms ? item.started_at + item.duration_ms : now - let overlapCount = 0 - - for (const otherItem of allItems) { - if (otherItem.id === item.id || !otherItem.started_at) continue - - const otherEnd = otherItem.duration_ms ? otherItem.started_at + otherItem.duration_ms : now - - // Check if time ranges overlap - if (item.started_at < otherEnd && otherItem.started_at < itemEnd) { - overlapCount++ - } + // Create events for start and end times + interface Event { + time: number + type: 'start' | 'end' + itemId: string } - // Base opacity of 1, reduce by 0.2 for each overlap, minimum 0.3 - return Math.max(0.3, 1 - overlapCount * 0.2) + const events: Event[] = [] + + for (const item of items) { + if (!item.started_at) continue + + const endTime = item.duration_ms ? item.started_at + item.duration_ms : now + events.push({ time: item.started_at, type: 'start', itemId: item.id }) + events.push({ time: endTime, type: 'end', itemId: item.id }) + overlapCounts.set(item.id, 0) + } + + // Sort events by time, with end events before start events at the same time + events.sort((a, b) => { + if (a.time !== b.time) return a.time - b.time + return a.type === 'end' ? -1 : 1 + }) + + // Sweep through events + const activeItems = new Set() + + for (const event of events) { + if (event.type === 'start') { + // Count current active items as overlaps for this item + overlapCounts.set(event.itemId, activeItems.size) + + // Update overlap counts for all currently active items + for (const activeId of activeItems) { + overlapCounts.set(activeId, overlapCounts.get(activeId)! + 1) + } + + activeItems.add(event.itemId) + } else { + activeItems.delete(event.itemId) + } + } + return Object.fromEntries(overlapCounts.entries()) } + + // At component level, compute once when items change + const allOverlaps = $derived(computeAllOverlaps(filteredItems)) + const maximumOverlaps = $derived(Math.max(...Object.values(allOverlaps))) + const opacity = $derived(Math.max(0.02, 1 / maximumOverlaps)) + // Then in your template, use: + // allOverlaps.get(item.id) ?? 0 {#if min && filteredItems.length > 0 && startItem?.started_at} @@ -155,24 +186,6 @@ {/if}
- {:else if hasMoreIterations} - - - {#snippet text()} - Load previous iterations - {/snippet} - {:else}
{/if} @@ -205,7 +218,8 @@ {#each filteredItems as item, i} {#if item.started_at} {@const position = calculateItemPosition(item)} - {@const opacity = getOverlapOpacity(item, filteredItems)} + { - scheduledFor = response }) + .then((response) => { + scheduledFor = response + }) + .catch((error) => { + console.error('Failed to fetch scheduled for:', error) + }) } catch (error) { console.error('Failed to fetch scheduled for:', error) } diff --git a/frontend/src/lib/components/flows/flowModuleNextId.ts b/frontend/src/lib/components/flows/flowModuleNextId.ts index 428ceccf47..1d0cd14ec6 100644 --- a/frontend/src/lib/components/flows/flowModuleNextId.ts +++ b/frontend/src/lib/components/flows/flowModuleNextId.ts @@ -6,6 +6,8 @@ import { charsToNumber, numberToChars } from './idUtils' // Computes the next available id export function nextId(flowState: FlowState, fullFlow: OpenFlow): string { const allIds = dfs(fullFlow.value.modules, (fm) => fm.id) + console.log('allIds', allIds) + const max = allIds.concat(Object.keys(flowState)).reduce((acc, key) => { if (key.length >= 4) { return acc diff --git a/frontend/src/lib/components/graph/model.ts b/frontend/src/lib/components/graph/model.ts index a97f603617..8ed90faeeb 100644 --- a/frontend/src/lib/components/graph/model.ts +++ b/frontend/src/lib/components/graph/model.ts @@ -27,11 +27,6 @@ export type DurationStatus = { byJob: Record } -export type GlobalIterationBounds = { - iteration_from?: number - iteration_total?: number -} - export type FlowStatusViewerContext = { flowStateStore?: FlowState retryStatus: StateStore> @@ -58,6 +53,10 @@ export type GraphModuleState = { selectedForloopIndex?: number selectedForLoopSetManually?: boolean flow_jobs_success?: (boolean | undefined)[] + flow_jobs_duration?: { + started_at?: (string | undefined)[] + duration_ms?: (number | undefined)[] + } flow_jobs?: string[] iteration_total?: number retries?: number diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index 346ed62d1c..09feda8a6c 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -89,7 +89,7 @@ components: minimum: 0 maximum: 100 retry_if: - $ref: '#/components/schemas/RetryIf' + $ref: "#/components/schemas/RetryIf" RetryIf: type: object @@ -569,6 +569,17 @@ components: type: array items: type: boolean + flow_jobs_duration: + type: object + properties: + started_at: + type: array + items: + type: string + duration_ms: + type: array + items: + type: integer branch_chosen: type: object properties: