From 715d09c30dbf5196fc7ac8d524951ff61ef775fa Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 4 Dec 2024 18:24:31 +0100 Subject: [PATCH] simplify warn_after_seconds --- backend/windmill-common/src/utils.rs | 11 ++++++---- .../windmill-worker/src/result_processor.rs | 4 ++-- backend/windmill-worker/src/worker.rs | 20 +++++++++---------- backend/windmill-worker/src/worker_flow.rs | 8 ++++---- 4 files changed, 23 insertions(+), 20 deletions(-) diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index dc76c199e6..5dd89f80e1 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -34,7 +34,7 @@ pub const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); use crate::CRITICAL_ALERT_MUTE_UI_ENABLED; -use std::panic::{self, AssertUnwindSafe}; +use std::panic::{self, AssertUnwindSafe, Location}; use std::sync::atomic::Ordering; use crate::worker::CLOUD_HOSTED; @@ -544,13 +544,16 @@ use pin_project_lite::pin_project; pub trait WarnAfterExt: Future + Sized { /// Warns if the future takes longer than the specified number of seconds to complete. - fn warn_after_seconds(self, seconds: u8, location: &'static str) -> WarnAfterFuture { + #[track_caller] + fn warn_after_seconds(self, seconds: u8) -> WarnAfterFuture { + let caller = Location::caller(); + let location = format!("{}:{}", caller.file(), caller.line()); WarnAfterFuture { future: self, timeout: time::sleep(Duration::from_secs(seconds as u64)), warned: false, start_time: std::time::Instant::now(), - location, + location: location, seconds, } } @@ -567,7 +570,7 @@ pin_project! { #[pin] timeout: Sleep, warned: bool, - location: &'static str, + location: String, start_time: std::time::Instant, seconds: u8, } diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index 6d7d6b1acf..25bfade2e8 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -432,7 +432,7 @@ pub async fn process_completed_job( #[cfg(feature = "benchmark")] bench, ) - .warn_after_seconds(10, "update_flow_status_after_job_completion success") + .warn_after_seconds(10) .await?; } } @@ -471,7 +471,7 @@ pub async fn process_completed_job( #[cfg(feature = "benchmark")] bench, ) - .warn_after_seconds(10, "update_flow_status_after_job_completion error") + .warn_after_seconds(10) .await?; } } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 51c0c64fe1..f0fdc93885 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -160,7 +160,7 @@ pub async fn create_token_for_owner_in_bg( &email, &job_id, ) - .warn_after_seconds(5, "creating token for owner") + .warn_after_seconds(5) .await .expect("could not create job token"); *locked = token; @@ -1826,7 +1826,7 @@ async fn handle_queued_job( let daily_count = sqlx::query!( "SELECT value FROM metrics WHERE id = 'email_trigger_usage' AND created_at > NOW() - INTERVAL '1 day' ORDER BY created_at DESC LIMIT 1" ).fetch_optional(db) - .warn_after_seconds(5, "getting email_trigger_usage") + .warn_after_seconds(5) .await?.map(|x| serde_json::from_value::(x.value).unwrap_or(1)); if let Some(count) = daily_count { @@ -1840,7 +1840,7 @@ async fn handle_queued_job( serde_json::json!(count + 1) ) .execute(db) - .warn_after_seconds(5, "updating email_trigger_usage") + .warn_after_seconds(5) .await?; } } else { @@ -1848,7 +1848,7 @@ async fn handle_queued_job( "INSERT INTO metrics (id, value) VALUES ('email_trigger_usage', to_jsonb(1))" ) .execute(db) - .warn_after_seconds(5, "inserting email_trigger_usage") + .warn_after_seconds(5) .await?; } } @@ -1861,7 +1861,7 @@ async fn handle_queued_job( .ok_or_else(|| Error::InternalErr(format!("expected parent job")))?, job.id, ) - .warn_after_seconds(5, "updating flow status in progress") + .warn_after_seconds(5) .await?; Some(r) @@ -1874,7 +1874,7 @@ async fn handle_queued_job( &job.workspace_id ) .execute(db) - .warn_after_seconds(5, "updating parent job started_at flow_status") + .warn_after_seconds(5) .await { tracing::error!("Could not update parent job started_at flow_status: {}", e); } @@ -1891,7 +1891,7 @@ async fn handle_queued_job( job.workspace_id ) .fetch_one(db) - .warn_after_seconds(5, "getting job raw values") + .warn_after_seconds(5) .await .map(|record| (record.raw_code, record.raw_lock, record.raw_flow)) .unwrap_or_default(), @@ -1932,7 +1932,7 @@ async fn handle_queued_job( &job.parent_job.unwrap() ) .fetch_one(db) - .warn_after_seconds(5, "getting script path from queue for caching purposes") + .warn_after_seconds(5) .await .map_err(|e| { Error::InternalErr(format!( @@ -1967,7 +1967,7 @@ async fn handle_queued_job( &job.workspace_id, &cached_res_path, ) - .warn_after_seconds(5, "getting cached resource value") + .warn_after_seconds(5) .await; if let Some(cached_resource_value) = cached_resource_value_maybe { { @@ -2006,7 +2006,7 @@ async fn handle_queued_job( worker_dir, job_completed_tx.0.clone(), ) - .warn_after_seconds(10, "handling flow") + .warn_after_seconds(10) .await?; Ok(true) } else { diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index a62ee84319..1e4aefdba0 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1110,7 +1110,7 @@ pub async fn update_flow_status_after_job_completion_internal( worker_dir, job_completed_tx, ) - .warn_after_seconds(10, "handle_flow in update_flow_status") + .warn_after_seconds(10) .await { Err(err) => { @@ -1512,7 +1512,7 @@ pub async fn handle_flow( let schedule_path = flow_job.schedule_path.as_ref().unwrap(); let schedule = - get_schedule_opt(&mut tx, &flow_job.workspace_id, schedule_path).warn_after_seconds(5, "get schedule_opt in handle_flow").await?; + get_schedule_opt(&mut tx, &flow_job.workspace_id, schedule_path).warn_after_seconds(5).await?; tx.commit().await?; @@ -1524,7 +1524,7 @@ pub async fn handle_flow( flow_job.script_path.as_ref().unwrap(), &flow_job.workspace_id, ) - .warn_after_seconds(5, "handle_maybe_scheduled_job in handle_flow") + .warn_after_seconds(5) .await { match err { @@ -1552,7 +1552,7 @@ pub async fn handle_flow( worker_dir, job_completed_tx, ) - .warn_after_seconds(10, "push next flow job") + .warn_after_seconds(10) .await?; Ok(()) }