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

fix: add support for log compaction on docker jobs (#5732)

Browse Source 
* improve docker compact

* improve docker compact

* update ref

* update

* update agent workers
This commit is contained in:
Ruben Fiszel
2025-05-13 21:42:15 +02:00
committed by GitHub
parent 4662cfd67d
commit 95cf9c6814
5 changed files with 252 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
4dc1f25f4fcc013334d4cc1d07cbe60a22b56d1f
1c9466b1f7f737033eedb4ff8c89bebdfddd2f26
backend/windmill-worker/src/bash_executor.rs
+18 -1
View File
@@ -285,6 +285,8 @@ async fn handle_docker_job(
occupancy_metrics: &mut OccupancyMetrics,
killpill_rx: &mut tokio::sync::broadcast::Receiver<()>,
) -> Result<Box<RawValue>, 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);
backend/windmill-worker/src/handle_child.rs
+179 -144
View File
@@ -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::<bool>(false);
let (mut set_too_many_logs, mut too_many_logs) = watch::channel::<bool>(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<Item = io::Result<String>> + Send,
job_id: &Uuid,
w_id: &str,
worker: &str,
conn: &Connection,
set_too_many_logs: &mut watch::Sender<bool>,
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<u32>, nsjail: bool) -> i32 {
if pid.is_none() {
return -1;
backend/windmill-worker/src/job_logger.rs
+48 -7
View File
@@ -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<AtomicU32>,
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");
backend/windmill-worker/src/job_logger_ee.rs
+6 -6
View File
@@ -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<sqlx::Postgres>,
_logs: String,
_logs: &str,
_total_size: Arc<AtomicU32>,
_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<AtomicU32>,
_compact_kind: CompactLogs,
_worker_name: &str,