From df910d7441499f6b58f35bacfa987cce9379c7fa Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 14 Mar 2024 19:42:41 +0100 Subject: [PATCH] fix: be less agressive with log streaming for long jobs --- backend/windmill-common/src/job_metrics.rs | 11 ++++----- backend/windmill-common/src/lib.rs | 9 ++++--- backend/windmill-worker/src/common.rs | 28 ++++++++++++++++++++-- backend/windmill-worker/src/worker.rs | 10 +++++++- 4 files changed, 45 insertions(+), 13 deletions(-) diff --git a/backend/windmill-common/src/job_metrics.rs b/backend/windmill-common/src/job_metrics.rs index 65aacd5e35..f5b16d742c 100644 --- a/backend/windmill-common/src/job_metrics.rs +++ b/backend/windmill-common/src/job_metrics.rs @@ -141,7 +141,6 @@ pub async fn record_metric( } }; - let timestamp = chrono::Utc::now(); match metric_kind { MetricKind::ScalarInt => { sqlx::query!( @@ -163,22 +162,20 @@ pub async fn record_metric( } MetricKind::TimeseriesInt => { sqlx::query!( - "UPDATE job_stats SET timestamps = timestamps || $4, timeseries_int = timeseries_int || $5 WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", &workspace_id, &job_id, &metric_id, - &[timestamp], - &[value_int] + value_int ).execute(db).await?; } MetricKind::TimeseriesFloat => { sqlx::query!( - "UPDATE job_stats SET timestamps = timestamps || $4, timeseries_float = timeseries_float || $5 WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", &workspace_id, &job_id, &metric_id, - &[timestamp], - &[value_float] + value_float ).execute(db).await?; } } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 913b9446bd..65dcdd6e6e 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -83,14 +83,17 @@ pub async fn shutdown_signal( tokio::signal::unix::signal(SignalKind::terminate())? .recv() .await; - Ok(()) } #[cfg(any(target_os = "linux", target_os = "macos"))] tokio::select! { - _ = terminate() => {}, - _ = tokio::signal::ctrl_c() => {}, + _ = terminate() => { + tracing::info!("shutdown monitor received terminate"); + }, + _ = tokio::signal::ctrl_c() => { + tracing::info!("shutdown monitor received ctrl-c"); + }, _ = rx.recv() => { tracing::info!("shutdown monitor received killpill"); }, diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index c1da822cd1..f5b4c4b1f0 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -556,6 +556,9 @@ pub async fn update_job_poller( } tracing::info!("{worker_name}/{job_id} in {w_id} still running. mem: {current_mem}kB, peak mem: {mem_peak}kB"); + + let update_job_row = (!*SLOW_LOGS && (i < 20 || (i < 120 && i % 5 == 0) || i % 10 == 0)) || i % 20 == 0; + if update_job_row { #[cfg(feature = "enterprise")] { // tracking metric starting at i >= 2 b/c first point it useless and we don't want to track metric for super fast jobs @@ -594,6 +597,7 @@ pub async fn update_job_poller( }); break; } + } }, ); } @@ -622,7 +626,6 @@ pub async fn handle_child( sigterm: bool, ) -> error::Result<()> { let start = Instant::now(); - let write_logs_delay = Duration::from_millis(500); let pid = child.id(); #[cfg(target_os = "linux")] @@ -764,11 +767,25 @@ pub async fn handle_child( 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(write_logs_delay), do_write_.clone())) + .take_until(future::join(sleep(delay), do_write_.clone())) .boxed(); /* Read up until an error is encountered, @@ -976,6 +993,9 @@ pub fn lines_to_stream( lazy_static::lazy_static! { static ref RE_00: Regex = Regex::new('\u{00}'.to_string().as_str()).unwrap(); + pub static ref NO_LOGS: bool = std::env::var("NO_LOGS").ok().is_some_and(|x| x == "1" || x == "true"); + pub static ref 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 @@ -1260,6 +1280,10 @@ async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow< 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(), diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d900ba42df..f9759323ae 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -93,7 +93,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, + write_file, NO_LOGS, SLOW_LOGS, }, deno_executor::{generate_deno_lock, handle_deno_job}, go_executor::{handle_go_job, install_go_dependencies}, @@ -2715,6 +2715,14 @@ async fn handle_queued_job( &job.id, &worker_name, &job.tag )); + if *NO_LOGS { + logs.push_str("Logs are disabled for this worker\n"); + } + + if *SLOW_LOGS { + logs.push_str("Logs are 10x less frequent for this worker\n"); + } + #[cfg(not(feature = "enterprise"))] if job.concurrent_limit.is_some() { logs.push_str("---\n");