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

feat: refactor logging to improve performance by order of magnitude for longer jobs

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-16 17:55:28 +01:00
parent bbce38490a
commit 958df4f62c
27 changed files with 530 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-8a3340146fe2072c7d47984c2f768c1ef632f8146766b494398421a8fc552ab3.json → backend/.sqlx/query-04d9cb2edf6933a3b8efbe274e10227a37475ff4ec351577bfdade19a15596d4.json
Generated
+2 -2
View File
@@ -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"
}
backend/.sqlx/query-1bf2ca894246bd716875635b2d0c294a1ce2ed21916097ea165df240f7421a1e.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json → backend/.sqlx/query-2ade671449393541fa565088b21268dad137314d250f7ded502defb9a6de0b2f.json
Generated
+2 -3
View File
@@ -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"
}
backend/.sqlx/query-3dfa0bf34c38b5529f2bbad405d23bda84db69757e132859d7e04df38e503f3e.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-453501fbd61efd26647baf9b6ef702ce0bc2e920914e9f08fe5f2a5f4ab03f02.json
Generated
-24
View File
@@ -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"
}
backend/.sqlx/query-486f181a9ced2bdc7c8d93da22c9d3e229ef106174bff2c472dbe82622f382b6.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-5cb644f89a94b6e6a1d7a84155bafc460c1749946056ded8ca712dacecbf427d.json
Generated
+36
View File
@@ -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"
}
backend/.sqlx/query-ab3c979c20a8ba9f6d906193c3bc7bc1941f0ecbe7ee2d8a62d2e02faeaae3f0.json
Generated
+23
View File
@@ -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"
}
backend/.sqlx/query-c7dd35561e9b1cfd86238d410139f50e4d87c762e4867d8b32c11bd8c74846eb.json
Generated
+16
View File
@@ -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"
}
backend/.sqlx/query-6f246196df45be4f1860f04c0fb25b1076c9c3fa30d632f4a90cb37a8f4d419b.json → backend/.sqlx/query-d9b32f819c87250962dae44e34d13b234e765b4b90b961e7db494f4c544c08ca.json
Generated
+2 -2
View File
@@ -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"
}
backend/migrations/20240315143613_large_logs_table.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20240315143613_large_logs_table.up.sql
+8
View File
@@ -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
);
backend/windmill-api/src/jobs.rs
+95 -27
View File
@@ -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<Job> {
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<String> {
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<usize> {
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<String> {
match self {
Job::QueuedJob(job) => job.logs.clone(),
Job::CompletedJob(job) => job.logs.clone(),
}
}
pub fn flow_status(&self) -> Option<FlowStatus> {
match self {
Job::QueuedJob(job) => job
@@ -3198,9 +3247,11 @@ async fn get_job_update(
Query(JobUpdateQuery { running, log_offset }): Query<JobUpdateQuery>,
) -> error::JsonResult<JobUpdate> {
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<Response> {
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)
}
backend/windmill-common/src/worker.rs
+1
View File
@@ -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");
backend/windmill-queue/src/jobs.rs
+41 -10
View File
@@ -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<str>,
db: impl Borrow<Pool<Postgres>>,
) {
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<String>,
@@ -357,7 +393,6 @@ where
pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Send>(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
e: serde_json::Value,
@@ -396,7 +431,6 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
false,
false,
Json(&result),
logs,
mem_peak,
canceled_by,
rsmq,
@@ -438,7 +472,6 @@ pub async fn add_completed_job<
success: bool,
skipped: bool,
result: Json<&T>,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
rsmq: Option<R>,
@@ -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<Json<HashMap<String, Box<RawValue>>>>,
result as Json<&T>,
logs,
queued_job.raw_code,
queued_job.raw_lock,
canceled_by.is_some(),
backend/windmill-worker/nsjail/run.bash.config.proto
+17
View File
@@ -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}
backend/windmill-worker/nsjail/run.powershell.config.proto
+7
View File
@@ -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"
backend/windmill-worker/src/bash_executor.rs
+55 -38
View File
@@ -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<CanceledBy>,
job: &QueuedJob,
@@ -45,14 +44,17 @@ pub async fn handle_bash_job(
worker_name: &str,
envs: HashMap<String, String>,
) -> Result<Box<RawValue>, 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::<Vec<&str>>();
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<CanceledBy>,
job: &QueuedJob,
@@ -221,26 +230,26 @@ pub async fn handle_powershell_job(
.collect::<Vec<String>>();
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"
)))
}
backend/windmill-worker/src/bun_executor.rs
+28 -33
View File
@@ -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<CanceledBy>,
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<CanceledBy>,
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<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
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
}
backend/windmill-worker/src/common.rs
+21 -53
View File
@@ -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<Box<RawValue>> {
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<Postgres>) {
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<F, Fut>(
pub async fn handle_child(
job_id: &Uuid,
db: &Pool<Postgres>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by_ref: &mut Option<CanceledBy>,
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<R: tokio::io::AsyncBufRead + Unpin>(
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<str>, db: impl Borrow<Pool<Postgres>>) {
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()
backend/windmill-worker/src/dedicated_worker.rs
+7 -4
View File
@@ -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<Arc<QueuedJob>>,
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<QueuedJob> = jobs.pop_front().expect("pop");
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&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();
backend/windmill-worker/src/deno_executor.rs
+7 -18
View File
@@ -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<CanceledBy>,
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<String>,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job: &QueuedJob,
@@ -174,16 +171,8 @@ pub async fn handle_deno_job(
envs: HashMap<String, String>,
) -> error::Result<Box<RawValue>> {
// 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, String>, 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<Arc<QueuedJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
db: &sqlx::Pool<sqlx::Postgres>,
) -> Result<()> {
use windmill_common::variables;
@@ -537,6 +525,7 @@ for await (const chunk of Deno.stdin.readable) {{
token,
jobs_rx,
worker_name,
db,
)
.await
}
backend/windmill-worker/src/go_executor.rs
+29 -26
View File
@@ -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<CanceledBy>,
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<CanceledBy>,
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,
backend/windmill-worker/src/python_executor.rs
+16 -22
View File
@@ -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<CanceledBy>,
job_dir: &str,
@@ -103,8 +102,8 @@ pub async fn pip_compile(
worker_name: &str,
w_id: &str,
) -> error::Result<String> {
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<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -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<CanceledBy>,
) -> error::Result<Vec<String>> {
@@ -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<CanceledBy>,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -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<Arc<QueuedJob>>,
killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> error::Result<()> {
let mut logs = "".to_string();
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = 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
}
backend/windmill-worker/src/worker.rs
+66 -92
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
job: Arc::new(job),
success: true,
result: empty_result(),
logs: String::new(),
mem_peak: 0,
cached_res_path: None,
token: "".to_string(),
@@ -2156,6 +2155,7 @@ async fn spawn_dedicated_worker(
job_completed_tx,
dedicated_worker_rx,
killpill_rx,
&db,
)
.await
}
@@ -2237,16 +2237,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
// ) -> error::Result<()> {
pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Sync +
true,
false,
Json(&result),
logs,
mem_peak.to_owned(),
canceled_by,
rsmq.clone(),
@@ -2315,7 +2305,6 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
let result = add_completed_job_error(
db,
&job,
logs.to_string(),
mem_peak.to_owned(),
canceled_by,
serde_json::from_str(result.get()).unwrap_or_else(
@@ -2406,11 +2395,17 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
};
let rsmq_2 = rsmq.clone();
let update_job_future = || {
let update_job_future = || async {
append_logs(
job.id,
job.workspace_id.clone(),
format!("Unexpected error during job execution:\n{err:#?}"),
db,
)
.await;
add_completed_job_error(
db,
job,
format!("Unexpected error during job execution:\n{err:#?}"),
mem_peak,
canceled_by.clone(),
err.clone(),
@@ -2418,6 +2413,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
worker_name,
false,
)
.await
};
let update_job_future = if job.is_flow_step || job.is_flow() {
@@ -2458,10 +2454,16 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
get_queued_job(&parent_job_id, &job.workspace_id, &db).await
{
let e = json!({"message": err.to_string(), "name": "InternalErr"});
append_logs(
parent_job.id,
job.workspace_id.clone(),
format!("Unexpected error during flow job error handling:\n{err}"),
db,
)
.await;
let _ = add_completed_job_error(
db,
&parent_job,
format!("Unexpected error during flow job error handling:\n{err}"),
mem_peak,
canceled_by.clone(),
e,
@@ -2522,7 +2524,6 @@ pub enum SendResult {
pub struct JobCompleted {
pub job: Arc<QueuedJob>,
pub result: Box<RawValue>,
pub logs: String,
pub mem_peak: i32,
pub success: bool,
pub cached_res_path: Option<String>,
@@ -2532,7 +2533,6 @@ pub struct JobCompleted {
async fn do_nativets(
job: &QueuedJob,
logs: String,
client: &AuthedClientBackgroundTask,
code: String,
db: &Pool<Postgres>,
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
)
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
let mut mem_peak: i32 = 0;
let mut canceled_by: Option<CanceledBy> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
"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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
.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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<Box<RawValue>>,
job_dir: &str,
job_completed_tx: JobCompletedSender,
logs: String,
mem_peak: i32,
canceled_by: Option<CanceledBy>,
cached_res_path: Option<String>,
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<CanceledBy>,
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3398,6 +3380,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
.is_some_and(|y| y.to_string().as_str() == "true")
})
.unwrap_or(false);
let content = capture_dependency_job(
&job.id,
job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| {
@@ -3406,7 +3389,6 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
))
})?,
&raw_code,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3420,6 +3402,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
raw_deps,
)
.await;
match content {
Ok(content) => {
if job.script_hash.is_none() {
@@ -3460,6 +3443,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
if &job.language == &Some(ScriptLang::Python3) {
if let Ok(relative_imports) = parse_relative_imports(&raw_code, script_path) {
let mut logs = "".to_string();
logs.push_str("\n--- RELATIVE IMPORTS ---\n\n");
logs.push_str(&relative_imports.join("\n"));
if !relative_imports.is_empty() {
@@ -3487,6 +3471,7 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
}
tx.commit().await?;
}
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
if let Err(e) = trigger_python_dependents_to_recompute_dependencies(
w_id,
script_path,
@@ -3509,9 +3494,18 @@ async fn handle_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync + Clo
))
}
Err(error) => {
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -3631,7 +3624,6 @@ async fn handle_flow_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync
flow.modules = lock_modules(
flow.modules,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -3717,7 +3709,6 @@ fn get_deployment_msg_and_parent_path_from_args(
async fn lock_modules(
modules: Vec<FlowModule>,
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
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<CanceledBy>,
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: &QueuedJob,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -4051,7 +4033,6 @@ async fn handle_app_dependency_job<R: rsmq_async::RsmqConnection + Send + Sync +
let value = lock_modules_app(
value,
job,
logs,
mem_peak,
canceled_by,
job_dir,
@@ -4129,7 +4110,6 @@ async fn capture_dependency_job(
job_id: &Uuid,
job_language: &ScriptLang,
job_raw_code: &str,
logs: &mut String,
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
job_dir: &str,
@@ -4163,7 +4143,6 @@ async fn capture_dependency_job(
let req: std::result::Result<String, Error> = 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,
backend/windmill-worker/src/worker_flow.rs
+24 -17
View File
@@ -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<sqlx::Postgres>;
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
success,
skipped,
Json(&result),
logs,
0,
canceled_by,
rsmq.clone(),
frontend/src/lib/components/ParqetTableRenderer.svelte
+3 -2
View File
@@ -130,10 +130,11 @@
<DarkModeObserver bind:darkMode />
<div class={twMerge('mt-4 border shadow-sm divide-y flex flex-col h-full min-h-[600px]')}>
<div class="ag-theme-alpine h-full" class:ag-theme-alpine-dark={darkMode}>
<div bind:this={eGui} style="height:100%" />
<div class="ag-theme-alpine h-full" class:ag-theme-alpine-dark={darkMode} style="height: 600px;">
<div bind:this={eGui} style="height:100%; " />
</div>
</div>
<!-- <div class="flex gap-1 absolute bottom-1 right-2 text-sm text-secondary"
>{firstRow}{'->'}{lastRow + 1} of {datasource?.rowCount} rows</div
> -->