From d15b889fb4bd055d112dd75f31f02fb216e86e28 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 28 Jun 2025 19:02:34 +0200 Subject: [PATCH] warn after long list_jobs duration --- backend/windmill-api/src/jobs.rs | 59 +++------------------------- backend/windmill-common/src/utils.rs | 29 ++++++++++++-- 2 files changed, 32 insertions(+), 56 deletions(-) diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index c5b5fb14ad..c8a8700289 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -30,6 +30,7 @@ use windmill_common::auth::is_super_admin_email; use windmill_common::error::JsonResult; use windmill_common::flow_status::{JobResult, RestartedFrom}; use windmill_common::jobs::{format_completed_job_result, format_result, ENTRYPOINT_OVERRIDE}; +use windmill_common::utils::WarnAfterExt; use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR}; use windmill_common::scripts::PREVIEW_IS_CODEBASE_HASH; @@ -96,36 +97,6 @@ use windmill_queue::{ PushArgsOwned, PushIsolationLevel, }; -#[cfg(feature = "prometheus")] -type Histo = prometheus::Histogram; - -#[cfg(not(feature = "prometheus"))] -type Histo = (); - -#[cfg(feature = "prometheus")] -fn setup_list_jobs_debug_metrics() -> Option { - let api_list_jobs_query_duration = if METRICS_DEBUG_ENABLED.load(Ordering::Relaxed) - && METRICS_ENABLED.load(Ordering::Relaxed) - { - Some( - prometheus::register_histogram!(prometheus::HistogramOpts::new( - "api_list_jobs_query_duration", - "Duration of listing jobs (query)", - )) - .expect("register prometheus metric"), - ) - } else { - None - }; - - api_list_jobs_query_duration -} - -#[cfg(not(feature = "prometheus"))] -fn setup_list_jobs_debug_metrics() -> Option { - None -} - pub fn workspaced_service() -> Router { let cors = CorsLayer::new() .allow_methods([http::Method::GET, http::Method::POST]) @@ -136,8 +107,6 @@ pub fn workspaced_service() -> Router { let ce_headers = ServiceBuilder::new().layer(axum::middleware::from_fn(add_webhook_allowed_origin)); - let api_list_jobs_query_duration = setup_list_jobs_debug_metrics(); - Router::new() .route( "/run/f/*script_path", @@ -212,10 +181,7 @@ pub fn workspaced_service() -> Router { ) .route("/add_batch_jobs/:n", post(add_batch_jobs)) .route("/run/preview_flow", post(run_preview_flow_job)) - .route( - "/list", - get(list_jobs).layer(Extension(api_list_jobs_query_duration)), - ) + .route("/list", get(list_jobs)) .route( "/list_selected_job_groups", // We use post because sending a huge array as a query param can produce @@ -1926,7 +1892,6 @@ async fn list_jobs( Path(w_id): Path, Query(pagination): Query, Query(lq): Query, - Extension(_api_list_jobs_query_duration): Extension>, ) -> error::JsonResult> { check_scopes(&authed, || format!("jobs:listjobs"))?; @@ -1990,24 +1955,12 @@ async fn list_jobs( }; let mut tx: Transaction<'_, Postgres> = user_db.begin(&authed).await?; - #[cfg(feature = "prometheus")] - let start = Instant::now(); - - #[cfg(feature = "prometheus")] - if _api_list_jobs_query_duration.is_some() || true { - tracing::info!("list_jobs query: {}", sql); - } - - let jobs: Vec = sqlx::query_as(&sql).fetch_all(&mut *tx).await?; + let jobs: Vec = sqlx::query_as(&sql) + .fetch_all(&mut *tx) + .warn_after_seconds_with_sql(5, format!("list_jobs: {}", sql)) + .await?; tx.commit().await?; - #[cfg(feature = "prometheus")] - if let Some(api_list_jobs_query_duration) = _api_list_jobs_query_duration { - let duration = start.elapsed().as_secs_f64(); - api_list_jobs_query_duration.observe(duration); - tracing::info!("list_jobs query took {}s: {}", duration, sql); - } - Ok(Json(jobs.into_iter().map(From::from).collect())) } diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 6b539dc73f..8cd6f25d83 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -752,16 +752,31 @@ pub trait WarnAfterExt: Future + Sized { #[track_caller] fn warn_after_seconds(self, seconds: u8) -> WarnAfterFuture { let caller = Location::caller(); + self.build_from_caller(seconds, caller, None) + } + + fn build_from_caller( + self, + seconds: u8, + caller: &Location, + sql: Option, + ) -> WarnAfterFuture { 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, + sql, } } + #[track_caller] + fn warn_after_seconds_with_sql(self, seconds: u8, sql: String) -> WarnAfterFuture { + let caller = Location::caller(); + self.build_from_caller(seconds, caller, Some(sql)) + } } // Blanket implementation for all futures. @@ -778,6 +793,7 @@ pin_project! { location: String, start_time: std::time::Instant, seconds: u8, + sql: Option, } } @@ -787,13 +803,20 @@ impl Future for WarnAfterFuture { fn poll(self: Pin<&mut Self>, cx: &mut TContext<'_>) -> Poll { let this = self.project(); + fn build_query_string(location: &str, sql: Option<&str>) -> String { + match sql { + Some(sql) => format!("{}: {}", location, sql), + None => location.to_string(), + } + } + // Poll the timeout future to check if it has elapsed. if !*this.warned { if this.timeout.poll(cx).is_ready() { tracing::warn!( location = this.location, "SLOW_QUERY: query {} to db taking longer than expected (> {} seconds)", - this.location, + build_query_string(&this.location, this.sql.as_deref()), this.seconds, ); *this.warned = true; @@ -808,7 +831,7 @@ impl Future for WarnAfterFuture { tracing::warn!( location = this.location, "SLOW_QUERY: completed query {} with total duration: {:.2?}", - this.location, + build_query_string(&this.location, this.sql.as_deref()), elapsed ); }