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

add MAX_RESULT_SIZE for CLOUD_HOSTED

Browse Source
This commit is contained in:
Ruben Fiszel
2023-02-07 01:52:55 +01:00
parent ac991dddbc
commit 3f8916cbc2
2 changed files with 41 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/jobs.rs
+2 -2
View File
@@ -474,7 +474,7 @@ async fn list_jobs(
"running",
"script_hash",
"script_path",
"args",
"CASE WHEN pg_column_size(args) > 1000 THEN '\"too large args\"'::jsonb ELSE args END",
"null as duration_ms",
"null as success",
"false as deleted",
@@ -509,7 +509,7 @@ async fn list_jobs(
"null as running",
"script_hash",
"script_path",
"args",
"CASE WHEN pg_column_size(args) > 1000 THEN '\"too large args\"'::jsonb ELSE args END",
"duration_ms",
"success",
"deleted",
backend/windmill-worker/src/worker.rs
+39 -31
View File
@@ -26,7 +26,7 @@ use windmill_common::{
utils::rd_string,
variables,
};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob, CLOUD_HOSTED};
use serde_json::{json, Value};
@@ -316,6 +316,11 @@ lazy_static::lazy_static! {
);
}
//only matter if CLOUD_HOSTED
const MAX_RESULT_SIZE: usize = 1024 * 1024 * 2; // 2MB
#[tracing::instrument(level = "trace")]
pub async fn run_worker(
db: &Pool<Postgres>,
@@ -426,10 +431,10 @@ pub async fn run_worker(
.ok()
.map(|x| format!(";{x}"))
.unwrap_or_else(|| String::new());
let max_log_size = std::env::var("MAX_LOG_SIZE")
.ok()
.and_then(|x| x.parse::<i64>().ok())
.unwrap_or(500000);
let deno_flags = std::env::var("DENO_FLAGS")
.ok()
.map(|x| x.split(' ').map(|x| x.to_string()).collect());
@@ -469,7 +474,6 @@ pub async fn run_worker(
pip_index_url,
pip_extra_index_url,
pip_trusted_host,
max_log_size,
deno_flags,
deno_auth_tokens,
pip_local_dependencies,
@@ -789,7 +793,6 @@ struct Envs {
pip_trusted_host: Option<String>,
deno_auth_tokens: String,
deno_flags: Option<Vec<String>>,
max_log_size: i64,
pip_local_dependencies: Option<Vec<String>>,
additional_python_paths: Option<Vec<String>>,
}
@@ -1185,7 +1188,7 @@ mount {{
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_go_job(
WorkerConfig { base_internal_url, disable_nuser, disable_nsjail, base_url, .. }: &WorkerConfig,
Envs { nsjail_path, go_path, path_env, home_env, max_log_size, .. }: &Envs,
Envs { nsjail_path, go_path, path_env, home_env, .. }: &Envs,
logs: &mut String,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -1223,7 +1226,6 @@ async fn handle_go_job(
job_dir,
db,
timeout,
*max_log_size,
go_path,
true,
skip_go_mod,
@@ -1344,7 +1346,7 @@ func Run(req Req) (interface{{}}, error){{
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
handle_child(&job.id, db, logs, timeout, *max_log_size, build_go, false).await?;
handle_child(&job.id, db, logs, timeout, build_go, false).await?;
Command::new(nsjail_path)
.current_dir(job_dir)
@@ -1370,14 +1372,14 @@ func Run(req Req) (interface{{}}, error){{
.stderr(Stdio::piped())
.spawn()?
};
handle_child(&job.id, db, logs, timeout, *max_log_size, child, !disable_nsjail).await?;
handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?;
read_result(job_dir).await
}
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_bash_job(
WorkerConfig { base_internal_url, disable_nuser, disable_nsjail, base_url, .. }: &WorkerConfig,
Envs { nsjail_path, path_env, home_env, max_log_size, .. }: &Envs,
Envs { nsjail_path, path_env, home_env, .. }: &Envs,
logs: &mut String,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -1449,7 +1451,7 @@ async fn handle_bash_job(
.stderr(Stdio::piped())
.spawn()?
};
handle_child(&job.id, db, logs, timeout, *max_log_size, child, !disable_nsjail).await?;
handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?;
//for now bash jobs have an empty result object
Ok(serde_json::json!(logs
.lines()
@@ -1469,7 +1471,7 @@ fn capitalize(s: &str) -> String {
#[tracing::instrument(level = "trace", skip_all)]
async fn handle_deno_job(
WorkerConfig { base_internal_url, base_url, disable_nuser, disable_nsjail, .. }: &WorkerConfig,
Envs { nsjail_path, deno_path, path_env, max_log_size, deno_auth_tokens, deno_flags, .. }: &Envs,
Envs { nsjail_path, deno_path, path_env, deno_auth_tokens, deno_flags, .. }: &Envs,
logs: &mut String,
job: &QueuedJob,
db: &sqlx::Pool<sqlx::Postgres>,
@@ -1613,7 +1615,7 @@ run().catch(async (e) => {{
}
.instrument(trace_span!("create_deno_jail"))
.await?;
handle_child(&job.id, db, logs, timeout, *max_log_size, child, !disable_nsjail).await?;
handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?;
read_result(job_dir).await
}
@@ -1642,7 +1644,7 @@ lazy_static! {
async fn handle_python_job(
WorkerConfig { base_internal_url, base_url, disable_nuser, disable_nsjail, .. }: &WorkerConfig,
envs @ Envs {
nsjail_path, python_path, path_env, max_log_size, additional_python_paths, ..
nsjail_path, python_path, path_env, additional_python_paths, ..
}: &Envs,
requirements_o: Option<String>,
job_dir: &str,
@@ -1887,7 +1889,7 @@ mount {{
.spawn()?
};
handle_child(&job.id, db, logs, timeout, *max_log_size, child, !disable_nsjail).await?;
handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?;
read_result(job_dir).await
}
@@ -1903,6 +1905,10 @@ async fn read_result(job_dir: &str) -> error::Result<serde_json::Value> {
let mut file = File::open(format!("{job_dir}/result.json")).await?;
let mut content = "".to_string();
file.read_to_string(&mut content).await?;
if *CLOUD_HOSTED && content.len() > MAX_RESULT_SIZE {
return Err(Error::ExecutionErr("Result is too large for the cloud app (limit 2MB).
If using this script as part of the flow, use the shared folder to pass heavy data between steps.".to_owned()));
}
serde_json::from_str(&content)
.map_err(|e| Error::ExecutionErr(format!("Error parsing result: {e}")))
}
@@ -2083,7 +2089,7 @@ async fn generate_deno_lock(
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
timeout: i32,
Envs { deno_path, max_log_size, .. }: &Envs,
Envs { deno_path, .. }: &Envs,
) -> error::Result<String> {
let _ = write_file(job_dir, "main.ts", code).await?;
@@ -2101,7 +2107,7 @@ async fn generate_deno_lock(
.stderr(Stdio::piped())
.spawn()?;
handle_child(job_id, db, logs, timeout, *max_log_size, child).await?;
handle_child(job_id, db, logs, timeout, * child).await?;
let path_lock = format!("{job_dir}/lock.json");
let mut file = File::open(path_lock).await?;
@@ -2132,7 +2138,6 @@ async fn capture_dependency_job(
job_dir,
db,
timeout,
envs.max_log_size,
&envs.go_path,
false,
false,
@@ -2155,7 +2160,7 @@ async fn pip_compile(
pip_extra_index_url,
pip_index_url,
pip_trusted_host,
max_log_size,
pip_local_dependencies,
..
}: &Envs,
@@ -2193,7 +2198,7 @@ async fn pip_compile(
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
handle_child(job_id, db, logs, timeout, *max_log_size, child, false)
handle_child(job_id, db, logs, timeout, child, false)
.await
.map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?;
let path_lock = format!("{job_dir}/requirements.txt");
@@ -2215,7 +2220,6 @@ async fn install_go_dependencies(
job_dir: &str,
db: &sqlx::Pool<sqlx::Postgres>,
timeout: i32,
max_log_size: i64,
go_path: &str,
preview: bool,
skip_go_mod: bool,
@@ -2229,7 +2233,7 @@ async fn install_go_dependencies(
.stderr(Stdio::piped())
.spawn()?;
handle_child(job_id, db, logs, timeout, max_log_size, child, false).await?;
handle_child(job_id, db, logs, timeout, child, false).await?;
}
let child = Command::new(go_path)
.current_dir(job_dir)
@@ -2237,7 +2241,7 @@ async fn install_go_dependencies(
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
handle_child(job_id, db, logs, timeout, max_log_size, child, false)
handle_child(job_id, db, logs, timeout, child, false)
.await
.map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?;
@@ -2343,7 +2347,6 @@ async fn handle_child(
db: &Pool<Postgres>,
logs: &mut String,
timeout: i32,
max_log_size: i64,
mut child: Child,
nsjail: bool,
) -> error::Result<()> {
@@ -2445,8 +2448,13 @@ 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 {
usize::MAX
};
/* log_remaining is zero when output limit was reached */
let mut log_remaining = (max_log_size as usize).saturating_sub(logs.chars().count());
let mut log_remaining = max_log_size.saturating_sub(logs.chars().count());
let mut result = io::Result::Ok(());
let mut output = output;
/* `do_write` resolves the task, but does not contain the Result.
@@ -2477,7 +2485,7 @@ async fn handle_child(
tracing::info!(%job_id, "Too many logs lines for job {job_id}");
let _ = set_too_many_logs.send(true);
joined.push_str(&format!(
"Job logs or result reached character limit of {max_log_size}; killing job."
"Job logs or result reached character limit of {MAX_RESULT_SIZE}; killing job."
));
/* stop reading and drop our streams fairly quickly */
break;
@@ -2539,7 +2547,7 @@ async fn handle_child(
match wait_result {
_ if *too_many_logs.borrow() => Err(Error::ExecutionErr(format!(
"logs or result reached limit. Set MAX_LOG_SIZE higher (current: {max_log_size})"
"logs or result reached limit. (current max size: {MAX_RESULT_SIZE} characters)"
))),
Ok(Ok(status)) => {
if status.success() {
@@ -2752,7 +2760,7 @@ async fn handle_python_reqs(
pip_extra_index_url,
pip_trusted_host,
nsjail_path,
max_log_size,
..
}: &Envs,
job: &QueuedJob,
@@ -2847,7 +2855,7 @@ async fn handle_python_reqs(
.spawn()?
};
let child = handle_child(&job.id, db, logs, timeout, *max_log_size, child, false).await;
let child = handle_child(&job.id, db, logs, timeout, child, false).await;
tracing::info!(
worker_name = %worker_name,
job_id = %job.id,