diff --git a/backend/.sqlx/query-8a3340146fe2072c7d47984c2f768c1ef632f8146766b494398421a8fc552ab3.json b/backend/.sqlx/query-04d9cb2edf6933a3b8efbe274e10227a37475ff4ec351577bfdade19a15596d4.json similarity index 60% rename from backend/.sqlx/query-8a3340146fe2072c7d47984c2f768c1ef632f8146766b494398421a8fc552ab3.json rename to backend/.sqlx/query-04d9cb2edf6933a3b8efbe274e10227a37475ff4ec351577bfdade19a15596d4.json index 45659d874d..5085b2b037 100644 --- a/backend/.sqlx/query-8a3340146fe2072c7d47984c2f768c1ef632f8146766b494398421a8fc552ab3.json +++ b/backend/.sqlx/query-04d9cb2edf6933a3b8efbe274e10227a37475ff4ec351577bfdade19a15596d4.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT running, substr(logs, $1) as logs, mem_peak, \n CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status \n FROM queue WHERE workspace_id = $2 AND id = $3", + "query": "SELECT running, substr(concat(coalesce(queue.logs, ''), job_logs.logs), $1) as logs, mem_peak, \n CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status \n FROM queue\n LEFT JOIN job_logs ON job_logs.job_id = queue.id \n WHERE queue.workspace_id = $2 AND queue.id = $3", "describe": { "columns": [ { @@ -38,5 +38,5 @@ null ] }, - "hash": "8a3340146fe2072c7d47984c2f768c1ef632f8146766b494398421a8fc552ab3" + "hash": "04d9cb2edf6933a3b8efbe274e10227a37475ff4ec351577bfdade19a15596d4" } diff --git a/backend/.sqlx/query-1bf2ca894246bd716875635b2d0c294a1ce2ed21916097ea165df240f7421a1e.json b/backend/.sqlx/query-1bf2ca894246bd716875635b2d0c294a1ce2ed21916097ea165df240f7421a1e.json deleted file mode 100644 index dde72218df..0000000000 --- a/backend/.sqlx/query-1bf2ca894246bd716875635b2d0c294a1ce2ed21916097ea165df240f7421a1e.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET logs = $1 WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "1bf2ca894246bd716875635b2d0c294a1ce2ed21916097ea165df240f7421a1e" -} diff --git a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json b/backend/.sqlx/query-2ade671449393541fa565088b21268dad137314d250f7ded502defb9a6de0b2f.json similarity index 65% rename from backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json rename to backend/.sqlx/query-2ade671449393541fa565088b21268dad137314d250f7ded502defb9a6de0b2f.json index 03bd3e00c1..c349ba77e5 100644 --- a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json +++ b/backend/.sqlx/query-2ade671449393541fa565088b21268dad137314d250f7ded502defb9a6de0b2f.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , logs\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms", + "query": "INSERT INTO completed_job AS cj\n ( workspace_id\n , id\n , parent_job\n , created_by\n , created_at\n , started_at\n , duration_ms\n , success\n , script_hash\n , script_path\n , args\n , result\n , raw_code\n , raw_lock\n , canceled\n , canceled_by\n , canceled_reason\n , job_kind\n , schedule_path\n , permissioned_as\n , flow_status\n , raw_flow\n , is_flow_step\n , is_skipped\n , language\n , email\n , visible_to_owner\n , mem_peak\n , tag\n , priority\n )\n VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($25, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,$10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $26, $27, $28, $29, $30)\n ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms", "describe": { "columns": [ { @@ -24,7 +24,6 @@ "Jsonb", "Text", "Text", - "Text", "Bool", "Varchar", "Text", @@ -92,5 +91,5 @@ false ] }, - "hash": "254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44" + "hash": "2ade671449393541fa565088b21268dad137314d250f7ded502defb9a6de0b2f" } diff --git a/backend/.sqlx/query-3dfa0bf34c38b5529f2bbad405d23bda84db69757e132859d7e04df38e503f3e.json b/backend/.sqlx/query-3dfa0bf34c38b5529f2bbad405d23bda84db69757e132859d7e04df38e503f3e.json new file mode 100644 index 0000000000..cd60dffec3 --- /dev/null +++ b/backend/.sqlx/query-3dfa0bf34c38b5529f2bbad405d23bda84db69757e132859d7e04df38e503f3e.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) \n FROM completed_job \n LEFT JOIN job_logs ON job_logs.job_id = completed_job.id \n WHERE completed_job.id = $1 AND completed_job.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "concat", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "3dfa0bf34c38b5529f2bbad405d23bda84db69757e132859d7e04df38e503f3e" +} diff --git a/backend/.sqlx/query-453501fbd61efd26647baf9b6ef702ce0bc2e920914e9f08fe5f2a5f4ab03f02.json b/backend/.sqlx/query-453501fbd61efd26647baf9b6ef702ce0bc2e920914e9f08fe5f2a5f4ab03f02.json deleted file mode 100644 index 9bfde90c97..0000000000 --- a/backend/.sqlx/query-453501fbd61efd26647baf9b6ef702ce0bc2e920914e9f08fe5f2a5f4ab03f02.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT substr(logs, $1) as logs FROM completed_job WHERE workspace_id = $2 AND id = $3", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "logs", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Int4", - "Text", - "Uuid" - ] - }, - "nullable": [ - null - ] - }, - "hash": "453501fbd61efd26647baf9b6ef702ce0bc2e920914e9f08fe5f2a5f4ab03f02" -} diff --git a/backend/.sqlx/query-486f181a9ced2bdc7c8d93da22c9d3e229ef106174bff2c472dbe82622f382b6.json b/backend/.sqlx/query-486f181a9ced2bdc7c8d93da22c9d3e229ef106174bff2c472dbe82622f382b6.json deleted file mode 100644 index 4a2c68993b..0000000000 --- a/backend/.sqlx/query-486f181a9ced2bdc7c8d93da22c9d3e229ef106174bff2c472dbe82622f382b6.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET logs = concat(logs, $1::text) WHERE id = $2", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "486f181a9ced2bdc7c8d93da22c9d3e229ef106174bff2c472dbe82622f382b6" -} diff --git a/backend/.sqlx/query-5cb644f89a94b6e6a1d7a84155bafc460c1749946056ded8ca712dacecbf427d.json b/backend/.sqlx/query-5cb644f89a94b6e6a1d7a84155bafc460c1749946056ded8ca712dacecbf427d.json new file mode 100644 index 0000000000..dddedac81f --- /dev/null +++ b/backend/.sqlx/query-5cb644f89a94b6e6a1d7a84155bafc460c1749946056ded8ca712dacecbf427d.json @@ -0,0 +1,36 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT substr(concat(coalesce(completed_job.logs, ''), job_logs.logs), $1) as logs, mem_peak, \n CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status \n FROM completed_job \n LEFT JOIN job_logs ON job_logs.job_id = completed_job.id \n WHERE completed_job.workspace_id = $2 AND id = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "logs", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "mem_peak", + "type_info": "Int4" + }, + { + "ordinal": 2, + "name": "flow_status", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Int4", + "Text", + "Uuid" + ] + }, + "nullable": [ + null, + true, + null + ] + }, + "hash": "5cb644f89a94b6e6a1d7a84155bafc460c1749946056ded8ca712dacecbf427d" +} diff --git a/backend/.sqlx/query-ab3c979c20a8ba9f6d906193c3bc7bc1941f0ecbe7ee2d8a62d2e02faeaae3f0.json b/backend/.sqlx/query-ab3c979c20a8ba9f6d906193c3bc7bc1941f0ecbe7ee2d8a62d2e02faeaae3f0.json new file mode 100644 index 0000000000..fc37929689 --- /dev/null +++ b/backend/.sqlx/query-ab3c979c20a8ba9f6d906193c3bc7bc1941f0ecbe7ee2d8a62d2e02faeaae3f0.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT right(logs, 300) FROM job_logs WHERE job_id = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "right", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "ab3c979c20a8ba9f6d906193c3bc7bc1941f0ecbe7ee2d8a62d2e02faeaae3f0" +} diff --git a/backend/.sqlx/query-c7dd35561e9b1cfd86238d410139f50e4d87c762e4867d8b32c11bd8c74846eb.json b/backend/.sqlx/query-c7dd35561e9b1cfd86238d410139f50e4d87c762e4867d8b32c11bd8c74846eb.json new file mode 100644 index 0000000000..73a1911081 --- /dev/null +++ b/backend/.sqlx/query-c7dd35561e9b1cfd86238d410139f50e4d87c762e4867d8b32c11bd8c74846eb.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO job_logs (logs, job_id, workspace_id) VALUES ($1, $2, $3) ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, $1::text)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "c7dd35561e9b1cfd86238d410139f50e4d87c762e4867d8b32c11bd8c74846eb" +} diff --git a/backend/.sqlx/query-6f246196df45be4f1860f04c0fb25b1076c9c3fa30d632f4a90cb37a8f4d419b.json b/backend/.sqlx/query-d9b32f819c87250962dae44e34d13b234e765b4b90b961e7db494f4c544c08ca.json similarity index 64% rename from backend/.sqlx/query-6f246196df45be4f1860f04c0fb25b1076c9c3fa30d632f4a90cb37a8f4d419b.json rename to backend/.sqlx/query-d9b32f819c87250962dae44e34d13b234e765b4b90b961e7db494f4c544c08ca.json index ff9dd8e9e5..7e1120ae85 100644 --- a/backend/.sqlx/query-6f246196df45be4f1860f04c0fb25b1076c9c3fa30d632f4a90cb37a8f4d419b.json +++ b/backend/.sqlx/query-d9b32f819c87250962dae44e34d13b234e765b4b90b961e7db494f4c544c08ca.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT logs FROM completed_job WHERE id = $1 AND workspace_id = $2", + "query": "SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2", "describe": { "columns": [ { @@ -19,5 +19,5 @@ true ] }, - "hash": "6f246196df45be4f1860f04c0fb25b1076c9c3fa30d632f4a90cb37a8f4d419b" + "hash": "d9b32f819c87250962dae44e34d13b234e765b4b90b961e7db494f4c544c08ca" } diff --git a/backend/migrations/20240315143613_large_logs_table.down.sql b/backend/migrations/20240315143613_large_logs_table.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20240315143613_large_logs_table.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20240315143613_large_logs_table.up.sql b/backend/migrations/20240315143613_large_logs_table.up.sql new file mode 100644 index 0000000000..5e3d8fd3b8 --- /dev/null +++ b/backend/migrations/20240315143613_large_logs_table.up.sql @@ -0,0 +1,8 @@ +-- Add up migration script here +CREATE TABLE IF NOT EXISTS job_logs +( + job_id uuid PRIMARY KEY, + workspace_id VARCHAR(255), + created_at TIMESTAMP WITH TIME ZONE DEFAULT now(), + logs TEXT +); \ No newline at end of file diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 782fed3e5b..821488eafe 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -38,7 +38,7 @@ use hyper::{http, Request, StatusCode}; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use sql_builder::{prelude::*, quote, SqlBuilder}; use sqlx::types::JsonRawValue; -use sqlx::{query_scalar, types::Uuid, FromRow, Postgres, Transaction}; +use sqlx::{types::Uuid, FromRow, Postgres, Transaction}; use tower_http::cors::{Any, CorsLayer}; use urlencoding::encode; use windmill_audit::audit_ee::audit_log; @@ -61,8 +61,8 @@ use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED}; use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL}; use windmill_queue::{ - add_completed_job_error, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, - push, CanceledBy, DecodeQueries, PushArgs, PushIsolationLevel, + add_completed_job_error, append_logs, get_queued_job, get_result_by_id_from_running_flow, + job_is_complete, push, CanceledBy, DecodeQueries, PushArgs, PushIsolationLevel, }; #[cfg(feature = "prometheus")] @@ -545,11 +545,14 @@ async fn get_job( async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::Result { let cjob_maybe = sqlx::query_as::<_, CompletedJob>("SELECT - id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, - CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, right(logs, 20000000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, + id, completed_job.workspace_id, parent_job, created_by, completed_job.created_at, duration_ms, success, script_hash, script_path, + CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, + right(concat(coalesce(completed_job.logs, ''), job_logs.logs), 20000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, raw_lock, email, visible_to_owner, mem_peak, tag, priority - FROM completed_job WHERE id = $1 AND workspace_id = $2") + FROM completed_job + LEFT JOIN job_logs ON completed_job.id = job_logs.job_id + WHERE id = $1 AND completed_job.workspace_id = $2") .bind(job_id) .bind(workspace_id) .fetch_optional(db) @@ -559,12 +562,16 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R Ok(cjob) } else { let job_o = sqlx::query_as::<_, QueuedJob>( - "SELECT id, workspace_id, parent_job, created_by, created_at, started_at, scheduled_for, running, - script_hash, script_path, CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, right(logs, 20000000) as logs, raw_code, canceled, canceled_by, canceled_reason, last_ping, + "SELECT id, queue.workspace_id, parent_job, created_by, queue.created_at, started_at, scheduled_for, running, + script_hash, script_path, CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '{\"reason\": \"WINDMILL_TOO_BIG\"}'::jsonb END as args, + right(concat(coalesce(queue.logs, ''), job_logs.logs), 20000) as logs, + raw_code, canceled, canceled_by, canceled_reason, last_ping, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, - suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak, + suspend, suspend_until, same_worker, raw_lock, pre_run_error, email, visible_to_owner, mem_peak, root_job, leaf_jobs, tag, concurrent_limit, concurrency_time_window_s, timeout, flow_step_id, cache_ttl, priority - FROM queue WHERE id = $1 AND workspace_id = $2", + FROM queue + LEFT JOIN job_logs ON queue.id = job_logs.job_id + WHERE id = $1 AND queue.workspace_id = $2", ) .bind(job_id) .bind(workspace_id) @@ -581,7 +588,10 @@ async fn get_job_logs( Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { let text = sqlx::query_scalar!( - "SELECT logs FROM completed_job WHERE id = $1 AND workspace_id = $2", + "SELECT CONCAT(coalesce(completed_job.logs, ''), coalesce(job_logs.logs, '')) + FROM completed_job + LEFT JOIN job_logs ON job_logs.job_id = completed_job.id + WHERE completed_job.id = $1 AND completed_job.workspace_id = $2", id, w_id ) @@ -871,10 +881,16 @@ async fn cancel_all( let job_running = get_queued_job(&j.id, &w_id, &db).await?; if let Some(job_running) = job_running { + append_logs( + j.id, + w_id.clone(), + format!("canceled by {username}: cancel_all"), + db.clone(), + ) + .await; let add_job = add_completed_job_error( &db, &job_running, - format!("canceled by {username}: cancel_all"), job_running.mem_peak.unwrap_or(0), Some(CanceledBy { username: Some(username.to_string()), @@ -1617,6 +1633,39 @@ impl Job { .flatten(), } } + + pub fn append_to_logs(&mut self, logs: &str) { + match self { + Job::QueuedJob(job) => { + if let Some(ref mut l) = job.logs { + l.push_str(logs); + } else { + job.logs = Some(logs.to_string()); + } + } + Job::CompletedJob(job) => { + if let Some(ref mut l) = job.logs { + l.push_str(logs); + } else { + job.logs = Some(logs.to_string()); + } + } + } + } + + pub fn log_len(&self) -> Option { + match self { + Job::QueuedJob(job) => job.logs.as_ref().map(|l| l.len()), + Job::CompletedJob(job) => job.logs.as_ref().map(|l| l.len()), + } + } + + pub fn logs(&self) -> Option { + match self { + Job::QueuedJob(job) => job.logs.clone(), + Job::CompletedJob(job) => job.logs.clone(), + } + } pub fn flow_status(&self) -> Option { match self { Job::QueuedJob(job) => job @@ -3198,9 +3247,11 @@ async fn get_job_update( Query(JobUpdateQuery { running, log_offset }): Query, ) -> error::JsonResult { let record = sqlx::query!( - "SELECT running, substr(logs, $1) as logs, mem_peak, + "SELECT running, substr(concat(coalesce(queue.logs, ''), job_logs.logs), $1) as logs, mem_peak, CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status - FROM queue WHERE workspace_id = $2 AND id = $3", + FROM queue + LEFT JOIN job_logs ON job_logs.job_id = queue.id + WHERE queue.workspace_id = $2 AND queue.id = $3", log_offset, &w_id, &job_id @@ -3221,23 +3272,29 @@ async fn get_job_update( flow_status: record.flow_status, })) } else { - let logs = query_scalar!( - "SELECT substr(logs, $1) as logs FROM completed_job WHERE workspace_id = $2 AND id = \ - $3", + let record = sqlx::query!( + "SELECT substr(concat(coalesce(completed_job.logs, ''), job_logs.logs), $1) as logs, mem_peak, + CASE WHEN is_flow_step is true then NULL else flow_status END as flow_status + FROM completed_job + LEFT JOIN job_logs ON job_logs.job_id = completed_job.id + WHERE completed_job.workspace_id = $2 AND id = $3", log_offset, &w_id, &job_id ) .fetch_optional(&db) .await?; - let logs = not_found_if_none(logs, "Job Update", job_id.to_string())?; - Ok(Json(JobUpdate { - running: Some(false), - completed: Some(true), - new_logs: logs, - mem_peak: record.as_ref().map(|r| r.mem_peak).flatten(), - flow_status: record.and_then(|r| r.flow_status), - })) + if let Some(record) = record { + Ok(Json(JobUpdate { + running: Some(false), + completed: Some(true), + new_logs: record.logs, + mem_peak: record.mem_peak, + flow_status: record.flow_status, + })) + } else { + Err(error::Error::NotFound(format!("Job not found: {}", job_id))) + } } } @@ -3417,7 +3474,7 @@ async fn get_completed_job<'a>( Path((w_id, id)): Path<(String, Uuid)>, ) -> error::Result { let job_o = sqlx::query("SELECT id, workspace_id, parent_job, created_by, created_at, duration_ms, success, script_hash, script_path, - CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, right(logs, 20000000) as logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, + CASE WHEN args is null or pg_column_size(args) < 2000000 THEN args ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as args, CASE WHEN result is null or pg_column_size(result) < 2000000 THEN result ELSE '\"WINDMILL_TOO_BIG\"'::jsonb END as result, logs, deleted, raw_code, canceled, canceled_by, canceled_reason, job_kind, env_id, schedule_path, permissioned_as, flow_status, raw_flow, is_flow_step, language, started_at, is_skipped, raw_lock, email, visible_to_owner, mem_peak, tag, priority FROM completed_job WHERE id = $1 AND workspace_id = $2") .bind(id) @@ -3426,7 +3483,18 @@ async fn get_completed_job<'a>( .await?; let job = not_found_if_none(job_o, "Completed Job", id.to_string())?; - let response = Json(CompletedJob::from_row(&job)?).into_response(); + let cj = CompletedJob::from_row(&job)?; + tracing::error!("response: {:?}", cj.logs); + + let response = Json(cj).into_response(); + // let extra_log = query_scalar!( + // "SELECT substr(logs, $1) as logs FROM large_logs WHERE workspace_id = $2 AND job_id = $3", + // log_offset - len, + // &w_id, + // &job_id + // ) + // .fetch_optional(db) + // .await.ok().flatten().flatten(); Ok(response) } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7d073d69be..a767a8d672 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -13,6 +13,7 @@ use crate::{error, global_settings::CUSTOM_TAGS_SETTING, server::ServerConfig, D lazy_static::lazy_static! { pub static ref WORKER_GROUP: String = std::env::var("WORKER_GROUP").unwrap_or_else(|_| "default".to_string()); + pub static ref NO_LOGS: bool = std::env::var("NO_LOGS").ok().is_some_and(|x| x == "1" || x == "true"); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 311281f612..9b9d767973 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -7,6 +7,7 @@ */ use std::{ + borrow::Borrow, collections::{HashMap, HashSet}, sync::Arc, vec, @@ -61,7 +62,7 @@ use windmill_common::{ schedule::Schedule, scripts::{ScriptHash, ScriptLang}, users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, - worker::{to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, WORKER_CONFIG}, + worker::{to_raw_value, DEFAULT_TAGS_PER_WORKSPACE, NO_LOGS, WORKER_CONFIG}, DB, METRICS_ENABLED, }; @@ -163,10 +164,16 @@ pub async fn cancel_job<'c: 'async_recursion>( .clone() .unwrap_or_else(|| "unexplicited reasons".to_string()); let e = serde_json::json!({"message": format!("Job canceled: {reason} by {username}"), "name": "Canceled", "reason": reason, "canceler": username}); + append_logs( + id, + w_id.to_string(), + format!("canceled by {username}: (force cancel: {force_cancel})"), + db, + ) + .await; let add_job = add_completed_job_error( &db, &job_running, - format!("canceled by {username}: (force cancel: {force_cancel})"), job_running.mem_peak.unwrap_or(0), Some(CanceledBy { username: Some(username.to_string()), reason: Some(reason) }), e, @@ -216,6 +223,35 @@ pub async fn cancel_job<'c: 'async_recursion>( Ok((tx, Some(id))) } +/* TODO retry this? */ +#[tracing::instrument(level = "trace", skip_all)] +pub async fn append_logs( + job_id: uuid::Uuid, + workspace: String, + logs: impl AsRef, + db: impl Borrow>, +) { + if logs.as_ref().is_empty() { + return; + } + + if *NO_LOGS { + tracing::info!("NO LOGS [{job_id}]: {}", logs.as_ref()); + return; + } + if let Err(err) = sqlx::query!( + "INSERT INTO job_logs (logs, job_id, workspace_id) VALUES ($1, $2, $3) ON CONFLICT (job_id) DO UPDATE SET logs = concat(job_logs.logs, $1::text)", + logs.as_ref(), + job_id, + workspace, + ) + .execute(db.borrow()) + .await + { + tracing::error!(%job_id, %err, "error updating logs for large_log job {job_id}: {err}"); + } +} + pub async fn cancel_persistent_script_jobs<'c>( username: &str, reason: Option, @@ -357,7 +393,6 @@ where pub async fn add_completed_job_error( db: &Pool, queued_job: &QueuedJob, - logs: String, mem_peak: i32, canceled_by: Option, e: serde_json::Value, @@ -396,7 +431,6 @@ pub async fn add_completed_job_error, - logs: String, mem_peak: i32, canceled_by: Option, rsmq: Option, @@ -504,7 +537,6 @@ pub async fn add_completed_job< , script_path , args , result - , logs , raw_code , raw_lock , canceled @@ -524,9 +556,9 @@ pub async fn add_completed_job< , tag , priority ) - VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($26, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\ - $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $27, $28, $29, $30, $31) - ON CONFLICT (id) DO UPDATE SET success = $7, result = $11, logs = concat(cj.logs, $12) RETURNING duration_ms", + VALUES ($1, $2, $3, $4, $5, COALESCE($6, now()), COALESCE($25, (EXTRACT('epoch' FROM (now())) - EXTRACT('epoch' FROM (COALESCE($6, now()))))*1000), $7, $8, $9,\ + $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $26, $27, $28, $29, $30) + ON CONFLICT (id) DO UPDATE SET success = $7, result = $11 RETURNING duration_ms", queued_job.workspace_id, queued_job.id, queued_job.parent_job, @@ -538,7 +570,6 @@ pub async fn add_completed_job< queued_job.script_path, &queued_job.args as &Option>>>, result as Json<&T>, - logs, queued_job.raw_code, queued_job.raw_lock, canceled_by.is_some(), diff --git a/backend/windmill-worker/nsjail/run.bash.config.proto b/backend/windmill-worker/nsjail/run.bash.config.proto index 183f66537f..4ce891f98b 100644 --- a/backend/windmill-worker/nsjail/run.bash.config.proto +++ b/backend/windmill-worker/nsjail/run.bash.config.proto @@ -54,6 +54,9 @@ mount { rw: true } + + + mount { dst: "/tmp" fstype: "tmpfs" @@ -68,6 +71,13 @@ mount { mandatory: false } +mount { + src: "{JOB_DIR}/wrapper.sh" + dst: "/tmp/wrapper.sh" + is_bind: true + mandatory: false +} + mount { src: "/etc" dst: "/etc" @@ -100,6 +110,13 @@ mount { is_bind: true } +mount { + src: "{JOB_DIR}/result2.out" + dst: "/tmp/result2.out" + rw: true + is_bind: true +} + iface_no_lo: true {SHARED_MOUNT} diff --git a/backend/windmill-worker/nsjail/run.powershell.config.proto b/backend/windmill-worker/nsjail/run.powershell.config.proto index f9272138d4..c349850a39 100644 --- a/backend/windmill-worker/nsjail/run.powershell.config.proto +++ b/backend/windmill-worker/nsjail/run.powershell.config.proto @@ -68,6 +68,13 @@ mount { mandatory: false } +mount { + src: "{JOB_DIR}/wrapper.sh" + dst: "/tmp/wrapper.sh" + is_bind: true + mandatory: false +} + mount { src: "/etc" dst: "/etc" diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 0e0292b28b..a87b81b6d9 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -5,7 +5,7 @@ use serde_json::{json, value::RawValue}; use sqlx::types::Json; use tokio::process::Command; use windmill_common::{error::Error, jobs::QueuedJob, worker::to_raw_value}; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; const BIN_BASH: &str = "/bin/bash"; const NSJAIL_CONFIG_RUN_BASH_CONTENT: &str = include_str!("../nsjail/run.bash.config.proto"); @@ -19,7 +19,7 @@ lazy_static::lazy_static! { use crate::{ common::{ build_args_map, get_reserved_variables, handle_child, read_file, read_file_content, - set_logs, start_child_process, write_file, + start_child_process, write_file, }, AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NSJAIL_PATH, PATH_ENV, POWERSHELL_CACHE_DIR, POWERSHELL_PATH, TZ_ENV, @@ -32,7 +32,6 @@ lazy_static::lazy_static! { #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bash_job( - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -45,14 +44,17 @@ pub async fn handle_bash_job( worker_name: &str, envs: HashMap, ) -> Result, Error> { - logs.push_str("\n\n--- BASH CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + let logs1 = "\n\n--- BASH CODE EXECUTION ---\n".to_string(); + append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + + write_file(job_dir, "main.sh", &format!("set -e\n{content}")).await?; write_file( job_dir, - "main.sh", - &format!("set -e\n{content}\necho \"\"\nsleep 0.02"), + "wrapper.sh", + &format!("set -o pipefail\nset -e\nmkfifo bp\ncat bp | tail -1 > ./result2.out &\n /bin/bash ./main.sh \"$@\" 2>&1 | tee bp\nwait $!"), ) .await?; + let token = client.get_token().await; let mut reserved_variables = get_reserved_variables(job, &token, db).await?; reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); @@ -76,6 +78,7 @@ pub async fn handle_bash_job( let args = args_owned.iter().map(|s| &s[..]).collect::>(); let _ = write_file(job_dir, "result.json", "").await?; let _ = write_file(job_dir, "result.out", "").await?; + let _ = write_file(job_dir, "result2.out", "").await?; let child = if !*DISABLE_NSJAIL { let _ = write_file( @@ -87,7 +90,13 @@ pub async fn handle_bash_job( .replace("{SHARED_MOUNT}", shared_mount), ) .await?; - let mut cmd_args = vec!["--config", "run.config.proto", "--", "/bin/bash", "main.sh"]; + let mut cmd_args = vec![ + "--config", + "run.config.proto", + "--", + "/bin/bash", + "wrapper.sh", + ]; cmd_args.extend(args); let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str()); nsjail_cmd @@ -101,7 +110,7 @@ pub async fn handle_bash_job( .stderr(Stdio::piped()); start_child_process(nsjail_cmd, NSJAIL_PATH.as_str()).await? } else { - let mut cmd_args = vec!["main.sh"]; + let mut cmd_args = vec!["wrapper.sh"]; cmd_args.extend(&args); let mut bash_cmd = Command::new(BIN_BASH); bash_cmd @@ -120,7 +129,6 @@ pub async fn handle_bash_job( handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -148,13 +156,15 @@ pub async fn handle_bash_job( } } - //for now bash jobs have an empty result object - let last_line = serde_json::json!(logs - .lines() - .last() - .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) - .unwrap_or_else(String::new)); - Ok(to_raw_value(&last_line)) + let result_out_path2 = format!("{job_dir}/result2.out"); + if tokio::fs::metadata(&result_out_path2).await.is_ok() { + let result = read_file_content(&result_out_path2).await?; + return Ok(to_raw_value(&json!(result))); + } + + Ok(to_raw_value(&json!( + "No result.out, result2.out or result.json found" + ))) } fn raw_to_string(x: &str) -> String { @@ -166,7 +176,6 @@ fn raw_to_string(x: &str) -> String { } #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_powershell_job( - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -221,26 +230,26 @@ pub async fn handle_powershell_job( .collect::>(); let mut install_string: String = String::new(); + let mut logs1 = String::new(); for line in content.lines() { for cap in RE_POWERSHELL_IMPORTS.captures_iter(line) { let module = cap.get(1).unwrap().as_str(); if !installed_modules.contains(&module.to_lowercase()) { - logs.push_str(&format!("\n{} not found in cache", module.to_string())); + logs1.push_str(&format!("\n{} not found in cache", module.to_string())); // instead of using Install-Module, we use Save-Module so that we can specify the installation path install_string.push_str(&format!( "Save-Module -Path {} -Force {};", POWERSHELL_CACHE_DIR, module )); } else { - logs.push_str(&format!("\n{} found in cache", module.to_string())); + logs1.push_str(&format!("\n{} found in cache", module.to_string())); } } } - set_logs(logs, &job.id, db).await; if !install_string.is_empty() { - logs.push_str("\n\nInstalling modules..."); - set_logs(logs, &job.id, db).await; + logs1.push_str("\n\nInstalling modules..."); + append_logs(job.id, job.workspace_id.clone(), logs1, db).await; let child = Command::new("pwsh") .args(&["-Command", &install_string]) .stdout(Stdio::piped()) @@ -250,7 +259,6 @@ pub async fn handle_powershell_job( handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -264,8 +272,9 @@ pub async fn handle_powershell_job( .await?; } - logs.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + let mut logs2 = "".to_string(); + logs2.push_str("\n\n--- POWERSHELL CODE EXECUTION ---\n"); + append_logs(job.id, job.workspace_id.clone(), logs2, db).await; // make sure default (only allhostsallusers) modules are loaded, disable autoload (cache can be large to explore especially on cloud) and add /tmp/windmill/cache to PSModulePath let profile = format!( @@ -291,12 +300,19 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"", }; write_file(job_dir, "main.ps1", content.as_str()).await?; + write_file( + job_dir, + "wrapper.sh", + &format!("set -o pipefail\nset -e\nmkfifo bp\ncat bp | tail -1 > ./result2.out &\n{} -F ./main.ps1 \"$@\" 2>&1 | tee bp\nwait $!", POWERSHELL_PATH.as_str()), + ) + .await?; let token = client.get_token().await; let mut reserved_variables = get_reserved_variables(job, &token, db).await?; reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); let _ = write_file(job_dir, "result.json", "").await?; let _ = write_file(job_dir, "result.out", "").await?; + let _ = write_file(job_dir, "result2.out", "").await?; let child = if !*DISABLE_NSJAIL { let _ = write_file( @@ -313,9 +329,8 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"", "--config", "run.config.proto", "--", - POWERSHELL_PATH.as_str(), - "-F", - "main.ps1", + "/bin/bash", + "wrapper.sh", ]; cmd_args.extend(pwsh_args.iter().map(|x| x.as_str())); Command::new(NSJAIL_PATH.as_str()) @@ -330,9 +345,9 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"", .stderr(Stdio::piped()) .spawn()? } else { - let mut cmd_args = vec!["-F", "main.ps1"]; + let mut cmd_args = vec!["wrapper.sh"]; cmd_args.extend(pwsh_args.iter().map(|x| x.as_str())); - Command::new(POWERSHELL_PATH.as_str()) + Command::new("/bin/bash") .current_dir(job_dir) .env_clear() .envs(envs) @@ -349,7 +364,6 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"", handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -362,10 +376,13 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"", ) .await?; - let last_line = serde_json::json!(logs - .lines() - .last() - .map(|x| ANSI_ESCAPE_RE.replace_all(x, "").to_string()) - .unwrap_or_else(String::new)); - Ok(to_raw_value(&last_line)) + let result_out_path2 = format!("{job_dir}/result2.out"); + if tokio::fs::metadata(&result_out_path2).await.is_ok() { + let result = read_file_content(&result_out_path2).await?; + return Ok(to_raw_value(&json!(result))); + } + + Ok(to_raw_value(&json!( + "No result.out, result2.out or result.json found" + ))) } diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index 3c11018c9e..e98be33166 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -6,7 +6,7 @@ use regex::Regex; use serde_json::value::RawValue; use uuid::Uuid; use windmill_parser_ts::remove_pinned_imports; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; #[cfg(feature = "enterprise")] use crate::common::build_envs_map; @@ -14,8 +14,7 @@ use crate::common::build_envs_map; use crate::{ common::{ create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, - parse_npm_config, read_result, set_logs, start_child_process, write_file, - write_file_binary, + parse_npm_config, read_result, start_child_process, write_file, write_file_binary, }, AuthedClientBackgroundTask, BUNFIG_INSTALL_SCOPES, BUN_CACHE_DIR, BUN_PATH, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, NODE_PATH, NPM_CONFIG_REGISTRY, NPM_PATH, NSJAIL_PATH, PATH_ENV, @@ -53,7 +52,6 @@ lazy_static::lazy_static! { } pub async fn gen_lockfile( - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_id: &Uuid, @@ -110,7 +108,6 @@ pub async fn gen_lockfile( handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, @@ -124,10 +121,12 @@ pub async fn gen_lockfile( .await?; if trusted_deps.len() > 0 { - logs.push_str(&format!( + let logs1 = format!( "\ndetected trustedDependencies: {}\n", trusted_deps.join(", ") - )); + ); + append_logs(job_id.clone(), w_id.to_string(), logs1, db).await; + let mut content = "".to_string(); { let mut file = File::open(format!("{job_dir}/package.json")).await?; @@ -150,7 +149,6 @@ pub async fn gen_lockfile( } install_lockfile( - logs, mem_peak, canceled_by, job_id, @@ -224,7 +222,6 @@ registry = {} } pub async fn install_lockfile( - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_id: &Uuid, @@ -243,9 +240,13 @@ pub async fn install_lockfile( .args(vec!["install"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()); - if npm_mode { - logs.push_str("NPM mode\n") - } + + let mut npm_logs = if npm_mode { + "NPM mode\n".to_string() + } else { + "".to_string() + }; + let has_file = if npm_mode { let registry = NPM_CONFIG_REGISTRY.read().await.clone(); if let Some(registry) = registry { @@ -255,7 +256,7 @@ pub async fn install_lockfile( let mut splitted = registry.split(":_authToken="); let custom_registry = splitted.next().unwrap_or_default(); - logs.push_str(&format!( + npm_logs.push_str(&format!( "Using custom npm registry: {custom_registry} {}\n", if splitted.next().is_some() { "with authToken" @@ -274,13 +275,16 @@ pub async fn install_lockfile( false }; + if npm_mode { + append_logs(job_id.clone(), w_id.to_string(), npm_logs, db).await; + } + let child_process = start_child_process(child_cmd, &*BUN_PATH).await?; gen_bunfig(job_dir).await?; handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, @@ -337,7 +341,6 @@ fn get_annotation(inner_content: &str) -> Annotations { #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_bun_job( requirements_o: Option, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -392,7 +395,6 @@ pub async fn handle_bun_job( } install_lockfile( - logs, mem_peak, canceled_by, &job.id, @@ -410,10 +412,10 @@ pub async fn handle_bun_job( let trusted_deps = get_trusted_deps(inner_content); // if !*DISABLE_NSJAIL || !empty_trusted_deps || has_custom_config_registry { - logs.push_str("\n\n--- BUN INSTALL ---\n"); - set_logs(&logs, &job.id, &db).await; + let logs1 = "\n\n--- BUN INSTALL ---\n".to_string(); + append_logs(job.id, job.workspace_id.clone(), logs1, db).await; + let _ = gen_lockfile( - logs, mem_peak, canceled_by, &job.id, @@ -436,17 +438,14 @@ pub async fn handle_bun_job( let main_code = remove_pinned_imports(inner_content)?; let _ = write_file(job_dir, "main.ts", &main_code).await?; - if annotation.nodejs_mode { - logs.push_str("\n\n--- NODE CODE EXECUTION ---\n"); + let init_logs = if annotation.nodejs_mode { + "\n\n--- NODE CODE EXECUTION ---\n".to_string() } else { - logs.push_str("\n\n--- BUN CODE EXECUTION ---\n"); - } - - let logs_f = async { - set_logs(&logs, &job.id, &db).await; - Ok(()) as error::Result<()> + "\n\n--- BUN CODE EXECUTION ---\n".to_string() }; + append_logs(job.id.clone(), job.workspace_id.to_string(), init_logs, db).await; + let write_wrapper_f = async { // let mut start = Instant::now(); let args = @@ -583,10 +582,9 @@ plugin(p) } }; - let (reserved_variables, _, _, _) = tokio::try_join!( + let (reserved_variables, _, _) = tokio::try_join!( reserved_variables_args_out_f, write_wrapper_f, - logs_f, write_loader_f )?; @@ -604,7 +602,6 @@ plugin(p) handle_child( &job.id, db, - logs, mem_peak, canceled_by, child_process, @@ -730,7 +727,6 @@ plugin(p) handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -838,7 +834,6 @@ pub async fn start_worker( } install_lockfile( - &mut logs, &mut mem_peak, &mut canceled_by, &Uuid::nil(), @@ -856,7 +851,6 @@ pub async fn start_worker( let trusted_deps = get_trusted_deps(inner_content); logs.push_str("\n\n--- BUN INSTALL ---\n"); let _ = gen_lockfile( - &mut logs, &mut mem_peak, &mut canceled_by, &Uuid::nil(), @@ -977,6 +971,7 @@ plugin(p) token, jobs_rx, worker_name, + db, ) .await } diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 0b1fe72954..07dbd5defd 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -29,13 +29,12 @@ use windmill_common::{ }; use anyhow::Result; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; #[cfg(any(target_os = "linux", target_os = "macos"))] use std::os::unix::process::ExitStatusExt; use std::{ - borrow::Borrow, collections::{hash_map::DefaultHasher, HashMap}, hash::{Hash, Hasher}, io, panic, @@ -350,21 +349,6 @@ pub async fn read_result(job_dir: &str) -> error::Result> { return read_file(&format!("{job_dir}/result.json")).await; } -#[tracing::instrument(level = "trace", skip_all)] -pub async fn set_logs(logs: &str, id: &uuid::Uuid, db: &Pool) { - if sqlx::query!( - "UPDATE queue SET logs = $1 WHERE id = $2", - logs.to_owned(), - id - ) - .execute(db) - .await - .is_err() - { - tracing::error!(%id, "error updating logs for id {id}") - }; -} - pub fn capitalize(s: &str) -> String { let mut c = s.chars(); match c.next() { @@ -614,7 +598,6 @@ pub async fn update_job_poller( pub async fn handle_child( job_id: &Uuid, db: &Pool, - logs: &mut String, mem_peak: &mut i32, canceled_by_ref: &mut Option, mut child: Child, @@ -631,8 +614,17 @@ pub async fn handle_child( #[cfg(target_os = "linux")] if let Some(pid) = pid { //set the highest oom priority - let mut file = File::create(format!("/proc/{pid}/oom_score_adj")).await?; - let _ = file.write_all(b"1000").await; + if let Some(mut file) = File::create(format!("/proc/{pid}/oom_score_adj")) + .await + .map_err(|e| { + tracing::error!("Could not create oom_score_file to pid {pid}: {e}"); + e + }) + .ok() + { + let _ = file.write_all(b"1000").await; + let _ = file.sync_all().await; + } } else { tracing::info!("could not get child pid"); } @@ -667,8 +659,7 @@ pub async fn handle_child( let (timeout_duration, timeout_warn_msg) = resolve_job_timeout(&db, w_id, job_id, custom_timeout).await; if let Some(msg) = timeout_warn_msg { - logs.push_str(msg.as_str()); - append_logs(job_id, msg.as_str(), db).await; + append_logs(job_id, w_id.to_string(), msg.as_str(), db).await; } /* a future that completes when the child process exits */ @@ -745,6 +736,7 @@ pub async fn handle_child( /* a future that reads output from the child and appends to the database */ let lines = async move { + let max_log_size = if *CLOUD_HOSTED { MAX_RESULT_SIZE } else { @@ -753,12 +745,16 @@ pub async fn handle_child( /* log_remaining is zero when output limit was reached */ let mut log_remaining = if *CLOUD_HOSTED { - max_log_size.saturating_sub(logs.chars().count()) + max_log_size } else { usize::MAX }; let mut result = io::Result::Ok(()); - let mut output = output.take_until(rx2.recv()).boxed(); + let mut output = output.take_until(async { + let _ = rx2.recv().await; + //wait at most 50ms after end of a script for output stream to end + tokio::time::sleep(Duration::from_millis(50)).await; + }).boxed(); /* `do_write` resolves the task, but does not contain the Result. * It's useful to know if the task completed. */ let (mut do_write, mut write_result) = tokio::spawn(ready(())).remote_handle(); @@ -795,7 +791,6 @@ pub async fn handle_child( while let Some(line) = read_lines.next().await { match line { - Ok(_) if log_remaining == 0 => (), Ok(line) => { if line.is_empty() { continue; @@ -818,8 +813,6 @@ pub async fn handle_child( } } - logs.push_str(&joined); - /* Ensure the last flush completed before starting a new one. * @@ -837,7 +830,7 @@ pub async fn handle_child( panic::resume_unwind(p); } - (do_write, write_result) = tokio::spawn(append_logs(job_id, joined, db.clone())).remote_handle(); + (do_write, write_result) = tokio::spawn(append_logs(job_id, w_id.to_string(), joined, db.clone())).remote_handle(); if let Err(err) = result { tracing::error!(%job_id, %err, "error reading output for job {job_id}: {err}"); @@ -993,10 +986,8 @@ pub fn lines_to_stream( lazy_static::lazy_static! { static ref RE_00: Regex = Regex::new('\u{00}'.to_string().as_str()).unwrap(); - pub static ref NO_LOGS: bool = std::env::var("NO_LOGS").ok().is_some_and(|x| x == "1" || x == "true"); pub static ref NO_LOGS_AT_ALL: bool = std::env::var("NO_LOGS_AT_ALL").ok().is_some_and(|x| x == "1" || x == "true"); pub static ref SLOW_LOGS: bool = std::env::var("SLOW_LOGS").ok().is_some_and(|x| x == "1" || x == "true"); - } // as a detail, `BufReader::lines()` removes \n and \r\n from the strings it yields, // so this pushes \n to thd destination string in each call @@ -1277,29 +1268,6 @@ pub async fn save_in_cache( } } -/* TODO retry this? */ -#[tracing::instrument(level = "trace", skip_all)] -async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow>) { - if logs.as_ref().is_empty() { - return; - } - - if *NO_LOGS { - tracing::info!("NO LOGS [{job_id}]: {}", logs.as_ref()); - return; - } - if let Err(err) = sqlx::query!( - "UPDATE queue SET logs = concat(logs, $1::text) WHERE id = $2", - logs.as_ref(), - job_id, - ) - .execute(db.borrow()) - .await - { - tracing::error!(%job_id, %err, "error updating logs for job {job_id}: {err}"); - } -} - fn tentatively_improve_error(err: Error, executable: &str) -> Error { if err .to_string() diff --git a/backend/windmill-worker/src/dedicated_worker.rs b/backend/windmill-worker/src/dedicated_worker.rs index 7f1c926c2b..c3491ee618 100644 --- a/backend/windmill-worker/src/dedicated_worker.rs +++ b/backend/windmill-worker/src/dedicated_worker.rs @@ -8,7 +8,8 @@ use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, process::Command, }; -use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value}; +use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value, DB}; +use windmill_queue::append_logs; use std::{collections::VecDeque, process::Stdio, sync::Arc}; @@ -56,6 +57,7 @@ pub async fn handle_dedicated_process( token: &str, mut jobs_rx: Receiver>, worker_name: &str, + db: &DB, ) -> std::result::Result<(), error::Error> { //do not cache local dependencies let mut child = { @@ -147,15 +149,16 @@ pub async fn handle_dedicated_process( let job: Arc = jobs.pop_front().expect("pop"); match serde_json::from_str::>(&line.replace("wm_res[success]:", "").replace("wm_res[error]:", "")) { Ok(result) => { + append_logs(job.id, job.workspace_id.clone(), logs.clone(), db).await; if line.starts_with("wm_res[success]:") { - job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap() + job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap() } else { - job_completed_tx.send(JobCompleted { job , result, logs: logs, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap() + job_completed_tx.send(JobCompleted { job , result, mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap() } }, Err(e) => { tracing::error!("Could not deserialize job result `{line}`: {e:?}"); - job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap(); + job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), mem_peak: 0, canceled_by: None, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap(); }, }; logs = init_log.clone(); diff --git a/backend/windmill-worker/src/deno_executor.rs b/backend/windmill-worker/src/deno_executor.rs index ef45b777ab..7ac6912a82 100644 --- a/backend/windmill-worker/src/deno_executor.rs +++ b/backend/windmill-worker/src/deno_executor.rs @@ -3,12 +3,12 @@ use std::{collections::HashMap, process::Stdio}; use itertools::Itertools; use serde_json::value::RawValue; use uuid::Uuid; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; use crate::{ common::{ create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, - parse_npm_config, read_result, set_logs, start_child_process, write_file, + parse_npm_config, read_result, start_child_process, write_file, }, AuthedClientBackgroundTask, DENO_CACHE_DIR, DENO_PATH, DISABLE_NSJAIL, HOME_ENV, NPM_CONFIG_REGISTRY, PATH_ENV, TZ_ENV, @@ -85,7 +85,6 @@ async fn get_common_deno_proc_envs( pub async fn generate_deno_lock( job_id: &Uuid, code: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -138,7 +137,6 @@ pub async fn generate_deno_lock( handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, @@ -161,7 +159,6 @@ pub async fn generate_deno_lock( #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_deno_job( requirements_o: Option, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -174,16 +171,8 @@ pub async fn handle_deno_job( envs: HashMap, ) -> error::Result> { // let mut start = Instant::now(); - logs.push_str("\n\n--- DENO CODE EXECUTION ---\n"); - - let logs_to_set = logs.clone(); - let id = job.id.clone(); - let db2 = db.clone(); - - let set_logs_f = async { - set_logs(&logs_to_set, &id, &db2).await; - Ok(()) as error::Result<()> - }; + let logs1 = "\n\n--- DENO CODE EXECUTION ---\n".to_string(); + append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await; let main_override = get_main_override(job.args.as_ref()); @@ -267,8 +256,7 @@ try {{ Ok(reserved_variables) as error::Result<(HashMap, String)> }; - let (_, (reserved_variables, token), _, _, _) = tokio::try_join!( - set_logs_f, + let ((reserved_variables, token), _, _, _) = tokio::try_join!( reserved_variables_args_out_f, write_main_f, write_wrapper_f, @@ -342,7 +330,6 @@ try {{ handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -424,6 +411,7 @@ pub async fn start_worker( job_completed_tx: JobCompletedSender, jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, + db: &sqlx::Pool, ) -> Result<()> { use windmill_common::variables; @@ -537,6 +525,7 @@ for await (const chunk of Deno.stdin.readable) {{ token, jobs_rx, worker_name, + db, ) .await } diff --git a/backend/windmill-worker/src/go_executor.rs b/backend/windmill-worker/src/go_executor.rs index f003acc463..d7e1ac6cd3 100644 --- a/backend/windmill-worker/src/go_executor.rs +++ b/backend/windmill-worker/src/go_executor.rs @@ -14,12 +14,12 @@ use windmill_common::{ utils::calculate_hash, }; use windmill_parser_go::{parse_go_imports, REQUIRE_PARSE}; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; use crate::{ common::{ capitalize, create_args_and_out_file, get_reserved_variables, handle_child, read_result, - set_logs, start_child_process, write_file, + start_child_process, write_file, }, AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, GOPRIVATE, GOPROXY, GO_BIN_CACHE_DIR, GO_CACHE_DIR, HOME_ENV, NSJAIL_PATH, PATH_ENV, TZ_ENV, @@ -34,7 +34,6 @@ lazy_static::lazy_static! { #[tracing::instrument(level = "trace", skip_all)] pub async fn handle_go_job( - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job: &QueuedJob, @@ -74,13 +73,12 @@ pub async fn handle_go_job( }; if !bin_exists { - logs.push_str("\n\n--- GO DEPENDENCIES SETUP ---\n"); - set_logs(logs, &job.id, db).await; + let logs1 = "\n\n--- GO DEPENDENCIES SETUP ---\n".to_string(); + append_logs(job.id.clone(), job.workspace_id.to_string(), logs1, db).await; install_go_dependencies( &job.id, inner_content, - logs, mem_peak, canceled_by, job_dir, @@ -93,8 +91,9 @@ pub async fn handle_go_job( ) .await?; - logs.push_str("\n\n--- GO CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + let logs2 = "\n\n--- GO CODE EXECUTION ---\n".to_string(); + append_logs(job.id.clone(), job.workspace_id.to_string(), logs2, db).await; + create_args_and_out_file(client, job, job_dir, db).await?; { let sig = windmill_parser_go::parse_go_sig(&inner_content)?; @@ -197,7 +196,6 @@ func Run(req Req) (interface{{}}, error){{ handle_child( &job.id, db, - logs, mem_peak, canceled_by, build_go_process, @@ -211,20 +209,28 @@ func Run(req Req) (interface{{}}, error){{ .await?; create_dir(&bin_path).await?; - tokio::fs::copy(format!("{job_dir}/main"), format!("{bin_path}/main")).await?; - logs.push_str(&format!("write cached binary: {}\n", bin_path)); + let target = format!("{bin_path}/main"); + tokio::fs::copy(format!("{job_dir}/main"), &target).await?; + append_logs( + job.id.clone(), + job.workspace_id.to_string(), + format!("write cached binary: {}\n", bin_path), + db, + ) + .await; } else { let path = format!("{bin_path}/main"); - logs.push_str(&format!("found cached binary: {path}\n")); - tokio::fs::copy(&path, format!("{job_dir}/main")) - .await - .map_err(|e| { - Error::ExecutionErr(format!( - "could not copy cached binary from {path} to {job_dir}/main: {e:?}" - )) - })?; - logs.push_str("\n\n--- GO CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + let mut logs2 = "".to_string(); + logs2.push_str(&format!("found cached binary: {path}\n")); + let target = format!("{job_dir}/main"); + tokio::fs::symlink(&path, &target).await.map_err(|e| { + Error::ExecutionErr(format!( + "could not copy cached binary from {path} to {job_dir}/main: {e:?}" + )) + })?; + + logs2.push_str("\n\n--- GO CODE EXECUTION ---\n"); + append_logs(job.id.clone(), job.workspace_id.to_string(), logs2, db).await; create_args_and_out_file(client, job, job_dir, db).await?; } @@ -283,7 +289,6 @@ func Run(req Req) (interface{{}}, error){{ handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -322,7 +327,6 @@ use std::io::prelude::*; pub async fn install_go_dependencies( job_id: &Uuid, code: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -346,7 +350,6 @@ pub async fn install_go_dependencies( handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, @@ -389,7 +392,8 @@ pub async fn install_go_dependencies( .fetch_optional(db) .await? { - logs.push_str(&format!("\nfound cached resolution: {}", hash)); + let logs1 = format!("\nfound cached resolution: {}", hash); + append_logs(job_id.clone(), w_id.to_string(), logs1, db).await; gen_go_mod(code, job_dir, &cached).await?; skip_tidy = true; new_lockfile = false; @@ -411,7 +415,6 @@ pub async fn install_go_dependencies( handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 4d41aae993..fb3f43f77f 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -23,7 +23,7 @@ use windmill_common::{ #[cfg(feature = "enterprise")] use windmill_common::variables::get_secret_value_as_admin; -use windmill_queue::CanceledBy; +use windmill_queue::{append_logs, CanceledBy}; lazy_static::lazy_static! { static ref PYTHON_PATH: String = @@ -56,7 +56,7 @@ use windmill_common::s3_helpers::S3_CACHE_BUCKET; use crate::{ common::{ create_args_and_out_file, get_main_override, get_reserved_variables, handle_child, - read_result, set_logs, start_child_process, write_file, + read_result, start_child_process, write_file, }, AuthedClientBackgroundTask, DISABLE_NSJAIL, DISABLE_NUSER, HOME_ENV, HTTPS_PROXY, HTTP_PROXY, LOCK_CACHE_DIR, NO_PROXY, NSJAIL_PATH, PATH_ENV, PIP_CACHE_DIR, PIP_EXTRA_INDEX_URL, @@ -95,7 +95,6 @@ pub fn handle_ephemeral_token(x: String) -> String { pub async fn pip_compile( job_id: &Uuid, requirements: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -103,8 +102,8 @@ pub async fn pip_compile( worker_name: &str, w_id: &str, ) -> error::Result { + let mut logs = String::new(); logs.push_str(&format!("\nresolving dependencies...")); - set_logs(logs, job_id, db).await; logs.push_str(&format!("\ncontent of requirements:\n{}\n", requirements)); let requirements = if let Some(pip_local_dependencies) = WORKER_CONFIG.read().await.pip_local_dependencies.as_ref() @@ -192,10 +191,10 @@ pub async fn pip_compile( .stdout(Stdio::piped()) .stderr(Stdio::piped()); let child_process = start_child_process(child_cmd, "pip-compile").await?; + append_logs(job_id.clone(), w_id.to_string(), logs, db).await; handle_child( job_id, db, - logs, mem_peak, canceled_by, child_process, @@ -233,7 +232,6 @@ pub async fn handle_python_job( worker_dir: &str, worker_name: &str, job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, db: &sqlx::Pool, @@ -254,14 +252,18 @@ pub async fn handle_python_job( db, worker_name, worker_dir, - logs, mem_peak, canceled_by, ) .await?; - logs.push_str("\n\n--- PYTHON CODE EXECUTION ---\n"); - set_logs(logs, &job.id, db).await; + append_logs( + job.id.clone(), + job.workspace_id.to_string(), + "\n\n--- PYTHON CODE EXECUTION ---\n".to_string(), + db, + ) + .await; let ( import_loader, @@ -432,7 +434,6 @@ mount {{ handle_child( &job.id, db, - logs, mem_peak, canceled_by, child, @@ -627,7 +628,6 @@ async fn handle_python_deps( db: &DB, worker_name: &str, worker_dir: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, ) -> error::Result> { @@ -661,7 +661,6 @@ async fn handle_python_deps( pip_compile( job_id, &requirements, - logs, mem_peak, canceled_by, job_dir, @@ -685,7 +684,6 @@ async fn handle_python_deps( .collect(), job_id, w_id, - logs, mem_peak, canceled_by, db, @@ -707,7 +705,6 @@ pub async fn handle_python_reqs( requirements: Vec<&str>, job_id: &Uuid, w_id: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, db: &sqlx::Pool, @@ -793,8 +790,10 @@ pub async fn handle_python_reqs( } } - logs.push_str("\n--- PIP INSTALL ---\n"); - logs.push_str(&format!("\n{req} is being installed for the first time.\n It will be cached for all ulterior uses.")); + let mut logs1 = String::new(); + logs1.push_str("\n\n--- PIP INSTALL ---\n"); + logs1.push_str(&format!("\n{req} is being installed for the first time.\n It will be cached for all ulterior uses.")); + append_logs(job_id.clone(), w_id.to_string(), logs1, db).await; tracing::info!( worker_name = %worker_name, @@ -899,7 +898,6 @@ pub async fn handle_python_reqs( let child = handle_child( &job_id, db, - logs, mem_peak, canceled_by, child, @@ -963,7 +961,6 @@ pub async fn start_worker( jobs_rx: Receiver>, killpill_rx: tokio::sync::broadcast::Receiver<()>, ) -> error::Result<()> { - let mut logs = "".to_string(); let mut mem_peak: i32 = 0; let mut canceled_by: Option = None; let context = variables::get_reserved_variables( @@ -995,15 +992,11 @@ pub async fn start_worker( db, worker_name, job_dir, - &mut logs, &mut mem_peak, &mut canceled_by, ) .await?; - logs.push_str("\n\n--- PYTHON CODE EXECUTION ---\n"); - set_logs(&mut logs, &Uuid::nil(), db).await; - let _args = None; let ( import_loader, @@ -1123,6 +1116,7 @@ for line in sys.stdin: token, jobs_rx, worker_name, + db, ) .await } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 472977518c..ffb8b49571 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -8,7 +8,6 @@ use anyhow::Result; use const_format::concatcp; -use itertools::Itertools; #[cfg(feature = "prometheus")] use prometheus::{ core::{AtomicI64, GenericGauge}, @@ -46,14 +45,15 @@ use windmill_common::{ users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, utils::{rd_string, StripPath}, worker::{ - to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, WORKER_CONFIG, WORKER_GROUP, + to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, NO_LOGS, WORKER_CONFIG, + WORKER_GROUP, }, DB, IS_READY, }; use windmill_queue::{ - canceled_job_to_result, empty_result, get_queued_job, pull, push, CanceledBy, PushArgs, - PushIsolationLevel, WrappedError, HTTP_CLIENT, + append_logs, canceled_job_to_result, empty_result, get_queued_job, pull, push, CanceledBy, + PushArgs, PushIsolationLevel, WrappedError, HTTP_CLIENT, }; #[cfg(feature = "prometheus")] @@ -96,7 +96,7 @@ use crate::{ bun_executor::{gen_lockfile, get_trusted_deps, handle_bun_job}, common::{ build_args_map, get_cached_resource_value_if_valid, hash_args, read_result, save_in_cache, - write_file, NO_LOGS, NO_LOGS_AT_ALL, SLOW_LOGS, + write_file, NO_LOGS_AT_ALL, SLOW_LOGS, }, deno_executor::{generate_deno_lock, handle_deno_job}, go_executor::{handle_go_job, install_go_dependencies}, @@ -1586,7 +1586,6 @@ pub async fn run_worker( // ) -> error::Result<()> { pub async fn process_completed_job( - JobCompleted { - job, - result, - logs, - mem_peak, - success, - cached_res_path, - canceled_by, - .. - }: JobCompleted, + JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, .. }: JobCompleted, client: &AuthedClient, db: &DB, worker_dir: &str, @@ -2273,7 +2264,6 @@ pub async fn process_completed_job, pub result: Box, - pub logs: String, pub mem_peak: i32, pub success: bool, pub cached_res_path: Option, @@ -2532,7 +2533,6 @@ pub struct JobCompleted { async fn do_nativets( job: &QueuedJob, - logs: String, client: &AuthedClientBackgroundTask, code: String, db: &Pool, @@ -2560,7 +2560,7 @@ async fn do_nativets( &job.workspace_id, ) .await?; - Ok((result.0, [logs, result.1].join("\n\n"))) + Ok((result.0, result.1)) } #[derive(Deserialize, Serialize, Default)] @@ -2678,12 +2678,15 @@ async fn handle_queued_job( ) .await; if let Some(cached_resource_value) = cached_resource_value_maybe { - let logs = "Job skipped because args & path found in cache and not expired".to_string(); + { + let logs = + "Job skipped because args & path found in cache and not expired".to_string(); + append_logs(job.id, job.workspace_id.clone(), logs, db).await; + } job_completed_tx .send(JobCompleted { job: job, result: cached_resource_value, - logs, mem_peak: 0, canceled_by: None, success: true, @@ -2717,10 +2720,6 @@ async fn handle_queued_job( let mut mem_peak: i32 = 0; let mut canceled_by: Option = None; // println!("handle queue {:?}", SystemTime::now()); - if let Some(log_str) = &job.logs { - logs.push_str(&log_str); - logs.push_str("\n"); - } logs.push_str(&format!( "job {} on worker {} (tag: {})\n", @@ -2753,12 +2752,12 @@ async fn handle_queued_job( "handling job {}", job.id ); + append_logs(job.id, job.workspace_id.clone(), logs, db).await; let result = match job.job_kind { JobKind::Dependencies => { handle_dependency_job( &job, - &mut logs, &mut mem_peak, &mut canceled_by, job_dir, @@ -2773,7 +2772,6 @@ async fn handle_queued_job( } JobKind::FlowDependencies => handle_flow_dependency_job( &job, - &mut logs, &mut mem_peak, &mut canceled_by, job_dir, @@ -2788,7 +2786,6 @@ async fn handle_queued_job( .map(|()| serde_json::from_str("{}").unwrap()), JobKind::AppDependencies => handle_app_dependency_job( &job, - &mut logs, &mut mem_peak, &mut canceled_by, job_dir, @@ -2817,7 +2814,6 @@ async fn handle_queued_job( client, job_dir, worker_dir, - &mut logs, &mut mem_peak, &mut canceled_by, base_internal_url, @@ -2839,11 +2835,11 @@ async fn handle_queued_job( result, job_dir, job_completed_tx, - logs, mem_peak, canceled_by, cached_res_path, client.get_token().await, + db, ) .await?; }; @@ -2855,11 +2851,11 @@ async fn process_result( result: error::Result>, job_dir: &str, job_completed_tx: JobCompletedSender, - logs: String, mem_peak: i32, canceled_by: Option, cached_res_path: Option, token: String, + db: &DB, ) -> error::Result<()> { match result { Ok(r) => { @@ -2867,7 +2863,6 @@ async fn process_result( .send(JobCompleted { job: job, result: r, - logs, mem_peak, canceled_by, success: true, @@ -2885,17 +2880,16 @@ async fn process_result( if res.as_ref().is_some_and(|x| !x.get().is_empty()) { res.unwrap() } else { - let last_10_log_lines = logs - .lines() - .skip(logs.lines().count().max(13) - 13) - .join("\n") - .to_string() - .replace("\n\n", "\n"); + let last_10_log_lines = sqlx::query_scalar!( + "SELECT right(logs, 300) FROM job_logs WHERE job_id = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1", + &job.id, + &job.workspace_id + ).fetch_one(db).await.ok().flatten().unwrap_or("".to_string()); let log_lines = last_10_log_lines .split("CODE EXECUTION ---") .last() - .unwrap_or(&logs); + .unwrap_or(&last_10_log_lines); extract_error_value(log_lines, i) } @@ -2910,7 +2904,6 @@ async fn process_result( .send(JobCompleted { job: job, result: to_raw_value(&error_value), - logs: logs, mem_peak, canceled_by, success: false, @@ -3047,7 +3040,6 @@ async fn handle_code_execution_job( client: &AuthedClientBackgroundTask, job_dir: &str, worker_dir: &str, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, base_internal_url: &str, @@ -3180,24 +3172,21 @@ async fn handle_code_execution_job( ) .await; } else if language == Some(ScriptLang::Nativets) { - logs.push_str("\n--- FETCH TS EXECUTION ---\n"); + append_logs( + job.id, + job.workspace_id.clone(), + "\n--- FETCH TS EXECUTION ---\n", + db, + ) + .await; let code = format!( "const BASE_URL = '{base_internal_url}';\nconst WM_TOKEN = '{}';\n{}", &client.get_token().await, inner_content ); - let (result, ts_logs) = do_nativets( - job, - logs.clone(), - &client, - code, - db, - mem_peak, - canceled_by, - worker_name, - ) - .await?; - *logs = ts_logs; + let (result, ts_logs) = + do_nativets(job, &client, code, db, mem_peak, canceled_by, worker_name).await?; + append_logs(job.id, job.workspace_id.clone(), ts_logs, db).await; return Ok(result); } @@ -3248,7 +3237,6 @@ mount {{ worker_dir, worker_name, job, - logs, mem_peak, canceled_by, db, @@ -3263,7 +3251,6 @@ mount {{ Some(ScriptLang::Deno) => { handle_deno_job( requirements_o, - logs, mem_peak, canceled_by, job, @@ -3280,7 +3267,6 @@ mount {{ Some(ScriptLang::Bun) => { handle_bun_job( requirements_o, - logs, mem_peak, canceled_by, job, @@ -3297,7 +3283,6 @@ mount {{ } Some(ScriptLang::Go) => { handle_go_job( - logs, mem_peak, canceled_by, job, @@ -3315,7 +3300,6 @@ mount {{ } Some(ScriptLang::Bash) => { handle_bash_job( - logs, mem_peak, canceled_by, job, @@ -3332,7 +3316,6 @@ mount {{ } Some(ScriptLang::Powershell) => { handle_powershell_job( - logs, mem_peak, canceled_by, job, @@ -3366,7 +3349,6 @@ mount {{ #[tracing::instrument(level = "trace", skip_all)] async fn handle_dependency_job( job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -3398,6 +3380,7 @@ async fn handle_dependency_job { if job.script_hash.is_none() { @@ -3460,6 +3443,7 @@ async fn handle_dependency_job { + let logs2 = sqlx::query_scalar!( + "SELECT logs FROM job_logs WHERE job_id = $1 AND workspace_id = $2", + &job.id, + &job.workspace_id + ) + .fetch_optional(db) + .await? + .flatten() + .unwrap_or_else(|| "no logs".to_string()); sqlx::query!( "UPDATE script SET lock_error_logs = $1 WHERE hash = $2 AND workspace_id = $3", - &format!("{logs}\n{error}"), + &format!("{logs2}\n{error}"), &job.script_hash.unwrap_or(ScriptHash(0)).0, &job.workspace_id ) @@ -3605,7 +3599,6 @@ async fn trigger_python_dependents_to_recompute_dependencies< async fn handle_flow_dependency_job( job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -3631,7 +3624,6 @@ async fn handle_flow_dependency_job, job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -3754,7 +3745,6 @@ async fn lock_modules( modules: lock_modules( modules, job, - logs, mem_peak, canceled_by, job_dir, @@ -3777,7 +3767,6 @@ async fn lock_modules( b.modules = lock_modules( b.modules, job, - logs, mem_peak, canceled_by, job_dir, @@ -3799,7 +3788,6 @@ async fn lock_modules( b.modules = lock_modules( b.modules, job, - logs, mem_peak, canceled_by, job_dir, @@ -3816,7 +3804,6 @@ async fn lock_modules( let default = lock_modules( default, job, - logs, mem_peak, canceled_by, job_dir, @@ -3843,7 +3830,6 @@ async fn lock_modules( &job.id, &language, &content, - logs, mem_peak, canceled_by, job_dir, @@ -3878,7 +3864,6 @@ async fn lock_modules( path = path, language = ?language, error = ?error, - logs = ?logs, "Failed to generate flow lock for raw script" ); e.value = FlowModuleValue::RawScript { @@ -3903,7 +3888,6 @@ async fn lock_modules( async fn lock_modules_app( value: Value, job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -3929,12 +3913,12 @@ async fn lock_modules_app( .as_str() .unwrap_or_default() .to_string(); + let mut logs = "".to_string(); logs.push_str("Found lockable inline script. Generating lock...\n"); let new_lock = capture_dependency_job( &job.id, &language, &content, - logs, mem_peak, canceled_by, job_dir, @@ -3950,6 +3934,7 @@ async fn lock_modules_app( .await; match new_lock { Ok(new_lock) => { + append_logs(job.id, job.workspace_id.clone(), logs, db).await; v.insert( "lock".to_string(), serde_json::Value::String(new_lock), @@ -3976,7 +3961,6 @@ async fn lock_modules_app( lock_modules_app( b, job, - logs, mem_peak, canceled_by, job_dir, @@ -3999,7 +3983,6 @@ async fn lock_modules_app( lock_modules_app( b, job, - logs, mem_peak, canceled_by, job_dir, @@ -4021,7 +4004,6 @@ async fn lock_modules_app( async fn handle_app_dependency_job( job: &QueuedJob, - logs: &mut String, mem_peak: &mut i32, canceled_by: &mut Option, job_dir: &str, @@ -4051,7 +4033,6 @@ async fn handle_app_dependency_job, job_dir: &str, @@ -4163,7 +4143,6 @@ async fn capture_dependency_job( let req: std::result::Result = pip_compile( job_id, &reqs, - logs, mem_peak, canceled_by, job_dir, @@ -4178,7 +4157,6 @@ async fn capture_dependency_job( req.split("\n").filter(|x| !x.starts_with("--")).collect(), job_id, w_id, - logs, mem_peak, canceled_by, db, @@ -4190,9 +4168,8 @@ async fn capture_dependency_job( if let Err(e) = r { tracing::error!( - "Failed to install python dependencies to prefill the cache: {:?} \n{}", - e, - logs + "Failed to install python dependencies to prefill the cache: {:?} \n", + e ); } } @@ -4207,7 +4184,6 @@ async fn capture_dependency_job( install_go_dependencies( job_id, job_raw_code, - logs, mem_peak, canceled_by, job_dir, @@ -4229,7 +4205,6 @@ async fn capture_dependency_job( generate_deno_lock( job_id, job_raw_code, - logs, mem_peak, canceled_by, job_dir, @@ -4249,7 +4224,6 @@ async fn capture_dependency_job( vec![] }; let req = gen_lockfile( - logs, mem_peak, canceled_by, job_id, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index c4c9fad0fa..9568a022bf 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -43,8 +43,8 @@ use windmill_common::{ flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend}, }; use windmill_queue::{ - add_completed_job, add_completed_job_error, get_queued_job, handle_maybe_scheduled_job, - CanceledBy, PushIsolationLevel, WrappedError, + add_completed_job, add_completed_job_error, append_logs, get_queued_job, + handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError, }; type DB = sqlx::Pool; @@ -614,16 +614,18 @@ pub async fn update_flow_status_after_job_completion_internal< }; let done = if !should_continue_flow { - let logs = if flow_job.canceled { - "Flow job canceled\n".to_string() - } else if stop_early { - format!("Flow job stopped early because of a stop early predicate returning true\n") - } else if success { - "Flow job completed with success\n".to_string() - } else { - "Flow job completed with error\n".to_string() - }; - + { + let logs = if flow_job.canceled { + "Flow job canceled\n".to_string() + } else if stop_early { + format!("Flow job stopped early because of a stop early predicate returning true\n") + } else if success { + "Flow job completed with success\n".to_string() + } else { + "Flow job completed with error\n".to_string() + }; + append_logs(flow_job.id, w_id.to_string(), logs, db).await; + } #[cfg(feature = "enterprise")] if flow_job.parent_job.is_none() { // run the cleanup step only when the root job is complete @@ -646,7 +648,6 @@ pub async fn update_flow_status_after_job_completion_internal< add_completed_job_error( db, &flow_job, - logs, 0, Some(CanceledBy { username: flow_job.canceled_by.clone(), @@ -687,7 +688,6 @@ pub async fn update_flow_status_after_job_completion_internal< success, stop_early && skip_if_stop_early, Json(&nresult), - logs, 0, None, rsmq.clone(), @@ -705,7 +705,6 @@ pub async fn update_flow_status_after_job_completion_internal< |e| json!({"error": format!("Impossible to serialize error: {e}")}), ), ), - logs, 0, None, rsmq.clone(), @@ -731,10 +730,16 @@ pub async fn update_flow_status_after_job_completion_internal< { Err(err) => { let e = json!({"message": err.to_string(), "name": "InternalError"}); + append_logs( + flow_job.id, + w_id.to_string(), + format!("Unexpected error during flow chaining:\n{:#?}", e), + db, + ) + .await; let _ = add_completed_job_error( db, &flow_job, - "Unexpected error during flow chaining:\n".to_string(), 0, None, e, @@ -1511,7 +1516,10 @@ async fn push_next_flow_job let success = false; let skipped = false; + let logs = "Timed out waiting to be resumed".to_string(); + append_logs(flow_job.id, flow_job.workspace_id.clone(), logs.clone(), db).await; + let result = json!({ "error": {"message": logs, "name": "SuspendedTimeout"}}); let canceled_by = if flow_job.canceled { Some(CanceledBy { @@ -1527,7 +1535,6 @@ async fn push_next_flow_job success, skipped, Json(&result), - logs, 0, canceled_by, rsmq.clone(), diff --git a/frontend/src/lib/components/ParqetTableRenderer.svelte b/frontend/src/lib/components/ParqetTableRenderer.svelte index 599e5df3b3..7f2d25feb3 100644 --- a/frontend/src/lib/components/ParqetTableRenderer.svelte +++ b/frontend/src/lib/components/ParqetTableRenderer.svelte @@ -130,10 +130,11 @@
-
-
+
+
+