From 95cf9c68147bd018f75a31bc4dec68b47239cbba Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Tue, 13 May 2025 21:42:15 +0200 Subject: [PATCH] fix: add support for log compaction on docker jobs (#5732) * improve docker compact * improve docker compact * update ref * update * update agent workers --- backend/ee-repo-ref.txt | 2 +- backend/windmill-worker/src/bash_executor.rs | 19 +- backend/windmill-worker/src/handle_child.rs | 323 ++++++++++--------- backend/windmill-worker/src/job_logger.rs | 55 +++- backend/windmill-worker/src/job_logger_ee.rs | 12 +- 5 files changed, 252 insertions(+), 159 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 9600e384bf..e742bdd62f 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -4dc1f25f4fcc013334d4cc1d07cbe60a22b56d1f \ No newline at end of file +1c9466b1f7f737033eedb4ff8c89bebdfddd2f26 \ No newline at end of file diff --git a/backend/windmill-worker/src/bash_executor.rs b/backend/windmill-worker/src/bash_executor.rs index 7a7a78282c..9683c0d8d5 100644 --- a/backend/windmill-worker/src/bash_executor.rs +++ b/backend/windmill-worker/src/bash_executor.rs @@ -285,6 +285,8 @@ async fn handle_docker_job( occupancy_metrics: &mut OccupancyMetrics, killpill_rx: &mut tokio::sync::broadcast::Receiver<()>, ) -> Result, Error> { + use crate::job_logger::append_logs_with_compaction; + let client = bollard::Docker::connect_with_unix_defaults().map_err(to_anyhow)?; let container_id = job_id.to_string(); @@ -313,6 +315,7 @@ async fn handle_docker_job( let w_id = workspace_id.to_string(); let j_id = job_id.clone(); let conn2 = conn.clone(); + let worker_name2 = worker_name.to_string(); let (tx, mut rx) = tokio::sync::broadcast::channel::<()>(1); let mut killpill_rx = killpill_rx.resubscribe(); @@ -334,7 +337,21 @@ async fn handle_docker_job( log = log_stream.next() => { match log { Some(Ok(log)) => { - append_logs(&j_id, w_id.clone(), log.to_string(), &conn2).await; + match &conn2 { + Connection::Sql(db) => { + append_logs_with_compaction( + &j_id, + &w_id, + &log.to_string(), + &db, + &worker_name2, + ) + .await; + } + c @ Connection::Http(_) => { + append_logs(&j_id, &w_id, &log.to_string(), &c).await; + } + } } Some(Err(e)) => { tracing::error!("Error getting logs: {:?}", e); diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 6cf4272a58..d82a72df98 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -130,13 +130,13 @@ pub async fn handle_child( } else { tracing::info!("could not get child pid"); } - let (set_too_many_logs, mut too_many_logs) = watch::channel::(false); + let (mut set_too_many_logs, mut too_many_logs) = watch::channel::(false); let (tx, rx) = broadcast::channel::<()>(3); - let mut rx2 = tx.subscribe(); + let mut rx2: broadcast::Receiver<()> = tx.subscribe(); let output = child_joined_output_stream(&mut child, job_id.clone()); - let job_id = job_id.clone(); + let job_id: Uuid = job_id.clone(); /* the cancellation future is polled on by `wait_on_child` while * waiting for the child to exit normally */ @@ -296,147 +296,19 @@ 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 { - usize::MAX - }; - - /* log_remaining is zero when output limit was reached */ - let mut log_remaining = if *CLOUD_HOSTED { - max_log_size - } else { - usize::MAX - }; - let mut result = io::Result::Ok(()); - 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(); - - let mut log_total_size: u64 = 0; - let pg_log_total_size = Arc::new(AtomicU32::new(0)); - - let mut pipe_stdout = pipe_stdout; - - while let Some(line) = output.by_ref().next().await { - - let do_write_ = do_write.shared(); - - let delay = if start.elapsed() < Duration::from_secs(10) { - Duration::from_millis(500) - } else if start.elapsed() < Duration::from_secs(60){ - Duration::from_millis(2500) - } else { - Duration::from_millis(5000) - }; - - let delay = if *SLOW_LOGS { - delay * 10 - } else { - delay - }; - - let mut read_lines = stream::once(async { line }) - .chain(output.by_ref()) - /* after receiving a line, continue until some delay has passed - * _and_ the previous database write is complete */ - .take_until(future::join(sleep(delay), do_write_.clone())) - .boxed(); - - /* Read up until an error is encountered, - * handle log lines first and then the error... */ - let mut joined = String::new(); - - while let Some(line) = read_lines.next().await { - - match line { - Ok(line) => { - if line.is_empty() { - continue; - } - append_with_limit(&mut joined, &line, &mut log_remaining); - if log_remaining == 0 { - 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_RESULT_SIZE}; killing job." - )); - /* stop reading and drop our streams fairly quickly */ - break; - } - } - Err(err) => { - result = Err(err); - break; - } - } - } - - - /* Ensure the last flush completed before starting a new one. - * - * This shouldn't pause since `take_until()` reads lines until `do_write` - * resolves. We only stop reading lines before `take_until()` resolves if we reach - * EOF or a read error. In those cases, waiting on a database query to complete is - * fine because we're done. */ - - if let Some(Ok(p)) = do_write_ - .then(|()| write_result) - .await - .err() - .map(|err| err.try_into_panic()) - { - panic::resume_unwind(p); - } - - - let joined_len = joined.len() as u64; - log_total_size += joined_len; - let compact_logs = log_total_size > LARGE_LOG_THRESHOLD_SIZE as u64; - if compact_logs { - log_total_size = 0; - } - - let worker_name = worker.to_string(); - let w_id2 = w_id.to_string(); - - - if let Some(buf) = &mut pipe_stdout { - buf.push_str(&joined); - (do_write, write_result) = tokio::spawn(async { }).remote_handle(); - } else { - (do_write, write_result) = tokio::spawn(append_job_logs(job_id, w_id2, joined, conn.clone(), compact_logs, pg_log_total_size.clone(), worker_name)).remote_handle(); - } - - if let Err(err) = result { - tracing::error!(%job_id, %err, "error reading output for job {job_id} '{child_name}': {err}"); - break; - } - - if *set_too_many_logs.borrow() { - break; - } - } - - /* drop our end of the pipe */ - drop(output); - - if let Some(Ok(p)) = do_write - .then(|()| write_result) - .await - .err() - .map(|err| err.try_into_panic()) - { - panic::resume_unwind(p); - } - }.instrument(trace_span!("child_lines")); + let lines = write_lines( + output, + &job_id, + w_id, + worker, + conn, + &mut set_too_many_logs, + start, + pipe_stdout, + &mut rx2, + child_name, + ) + .instrument(trace_span!("child_lines")); let (wait_result, _) = tokio::join!(wait_on_child, lines); @@ -462,6 +334,169 @@ pub async fn handle_child( } } +pub async fn write_lines( + output: impl stream::Stream> + Send, + job_id: &Uuid, + w_id: &str, + worker: &str, + conn: &Connection, + set_too_many_logs: &mut watch::Sender, + start: Instant, + pipe_stdout: Option<&mut String>, + rx2: &mut broadcast::Receiver<()>, + child_name: &str, +) { + 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 = if *CLOUD_HOSTED { + max_log_size + } else { + usize::MAX + }; + let mut result = io::Result::Ok(()); + 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(); + + let mut log_total_size: u64 = 0; + let pg_log_total_size = Arc::new(AtomicU32::new(0)); + + let mut pipe_stdout = pipe_stdout; + + while let Some(line) = output.by_ref().next().await { + let do_write_ = do_write.shared(); + + let delay = if start.elapsed() < Duration::from_secs(10) { + Duration::from_millis(500) + } else if start.elapsed() < Duration::from_secs(60) { + Duration::from_millis(2500) + } else { + Duration::from_millis(5000) + }; + + let delay = if *SLOW_LOGS { delay * 10 } else { delay }; + + let mut read_lines = stream::once(async { line }) + .chain(output.by_ref()) + /* after receiving a line, continue until some delay has passed + * _and_ the previous database write is complete */ + .take_until(future::join(sleep(delay), do_write_.clone())) + .boxed(); + + /* Read up until an error is encountered, + * handle log lines first and then the error... */ + let mut joined = String::new(); + + let job_id = job_id.clone(); + while let Some(line) = read_lines.next().await { + match line { + Ok(line) => { + if line.is_empty() { + continue; + } + append_with_limit(&mut joined, &line, &mut log_remaining); + if log_remaining == 0 { + 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_RESULT_SIZE}; killing job." + )); + /* stop reading and drop our streams fairly quickly */ + break; + } + } + Err(err) => { + result = Err(err); + break; + } + } + } + + /* Ensure the last flush completed before starting a new one. + * + * This shouldn't pause since `take_until()` reads lines until `do_write` + * resolves. We only stop reading lines before `take_until()` resolves if we reach + * EOF or a read error. In those cases, waiting on a database query to complete is + * fine because we're done. */ + + if let Some(Ok(p)) = do_write_ + .then(|()| write_result) + .await + .err() + .map(|err| err.try_into_panic()) + { + panic::resume_unwind(p); + } + + let joined_len = joined.len() as u64; + log_total_size += joined_len; + let compact_logs = log_total_size > LARGE_LOG_THRESHOLD_SIZE as u64; + if compact_logs { + log_total_size = 0; + } + + let worker_name = worker.to_string(); + + if let Some(buf) = &mut pipe_stdout { + buf.push_str(&joined); + (do_write, write_result) = tokio::spawn(async {}).remote_handle(); + } else { + let conn = conn.clone(); + let worker_name = worker_name.to_string(); + let w_id = w_id.to_string(); + let job_id = job_id.clone(); + let pg_log_total_size = pg_log_total_size.clone(); + + (do_write, write_result) = tokio::spawn(async move { + append_job_logs( + &job_id, + &w_id, + &joined, + &conn, + compact_logs, + pg_log_total_size, + &worker_name, + ) + .await; + }) + .remote_handle(); + } + + if let Err(err) = result { + tracing::error!(%job_id, %err, "error reading output for job {job_id} '{child_name}': {err}"); + break; + } + + if *set_too_many_logs.borrow() { + break; + } + } + + /* drop our end of the pipe */ + drop(output); + + if let Some(Ok(p)) = do_write + .then(|()| write_result) + .await + .err() + .map(|err| err.try_into_panic()) + { + panic::resume_unwind(p); + } +} + pub(crate) async fn get_mem_peak(pid: Option, nsjail: bool) -> i32 { if pid.is_none() { return -1; diff --git a/backend/windmill-worker/src/job_logger.rs b/backend/windmill-worker/src/job_logger.rs index 97784340c5..f4abd1d0c2 100644 --- a/backend/windmill-worker/src/job_logger.rs +++ b/backend/windmill-worker/src/job_logger.rs @@ -1,8 +1,10 @@ use regex::Regex; pub use windmill_common::jobs::LARGE_LOG_THRESHOLD_SIZE; +use windmill_common::utils::WarnAfterExt; use windmill_common::worker::{Connection, CLOUD_HOSTED}; +use windmill_common::DB; use windmill_queue::append_logs; use std::sync::atomic::AtomicU32; @@ -26,23 +28,23 @@ pub enum CompactLogs { } pub async fn append_job_logs( - job_id: Uuid, - w_id: String, - logs: String, - conn: Connection, + job_id: &Uuid, + w_id: &str, + logs: &str, + conn: &Connection, must_compact_logs: bool, total_size: Arc, - worker_name: String, + worker_name: &str, ) -> () { match conn { Connection::Sql(db) if must_compact_logs => { #[cfg(all(feature = "enterprise", feature = "parquet"))] - s3_storage(job_id, &w_id, &db, logs, total_size, &worker_name).await; + s3_storage(&job_id, &w_id, &db, logs, total_size, worker_name).await; #[cfg(not(all(feature = "enterprise", feature = "parquet")))] { default_disk_log_storage( - job_id, + &job_id, &w_id, &db, logs, @@ -59,6 +61,45 @@ pub async fn append_job_logs( } } +pub async fn append_logs_with_compaction( + job_id: &Uuid, + w_id: &str, + logs: &str, + db: &DB, + worker_name: &str, +) { + let log_length = sqlx::query_scalar!( + "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) RETURNING length(logs)", + logs, + job_id, + &w_id, + ) + .fetch_one(db) + .warn_after_seconds(1) + .await; + match log_length { + Ok(length) => { + let len = length.unwrap_or(0); + let conn: Connection = db.into(); + if len > LARGE_LOG_THRESHOLD_SIZE as i32 { + append_job_logs( + &job_id, + w_id, + "", + &conn, + true, + Arc::new(AtomicU32::new(len as u32)), + worker_name, + ) + .await; + } + } + Err(err) => { + tracing::error!(%job_id, %err, "error updating logs for job {job_id}: {err}"); + } + } +} + lazy_static::lazy_static! { static ref RE_00: Regex = Regex::new('\u{00}'.to_string().as_str()).unwrap(); pub static ref NO_LOGS_AT_ALL: bool = std::env::var("NO_LOGS_AT_ALL").ok().is_some_and(|x| x == "1" || x == "true"); diff --git a/backend/windmill-worker/src/job_logger_ee.rs b/backend/windmill-worker/src/job_logger_ee.rs index 43cb43f673..4b1d34392c 100644 --- a/backend/windmill-worker/src/job_logger_ee.rs +++ b/backend/windmill-worker/src/job_logger_ee.rs @@ -9,22 +9,22 @@ use crate::job_logger::CompactLogs; #[cfg(all(feature = "enterprise", feature = "parquet"))] pub(crate) async fn s3_storage( - _job_id: Uuid, - _w_id: &String, + _job_id: &Uuid, + _w_id: &str, _db: &sqlx::Pool, - _logs: String, + _logs: &str, _total_size: Arc, - _worker_name: &String, + _worker_name: &str, ) { tracing::info!("Logs length of {_job_id} has exceeded a threshold. Implementation to store excess on s3 in not OSS"); } #[allow(dead_code)] pub(crate) async fn default_disk_log_storage( - job_id: Uuid, + job_id: &Uuid, _w_id: &str, _db: &DB, - _nlogs: String, + _logs: &str, _total_size: Arc, _compact_kind: CompactLogs, _worker_name: &str,