warn after long list_jobs duration

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<Histo> {
-    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<Histo> {
-    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<String>,
     Query(pagination): Query<Pagination>,
     Query(lq): Query<ListCompletedQuery>,
-    Extension(_api_list_jobs_query_duration): Extension<Option<Histo>>,
 ) -> error::JsonResult<Vec<Job>> {
     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<UnifiedJob> = sqlx::query_as(&sql).fetch_all(&mut *tx).await?;
+    let jobs: Vec<UnifiedJob> = 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()))
 }
 

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<Self> {
         let caller = Location::caller();
+        self.build_from_caller(seconds, caller, None)
+    }
+
+    fn build_from_caller(
+        self,
+        seconds: u8,
+        caller: &Location,
+        sql: Option<String>,
+    ) -> WarnAfterFuture<Self> {
         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<Self> {
+        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<String>,
     }
 }
 
@@ -787,13 +803,20 @@ impl<F: Future> Future for WarnAfterFuture<F> {
     fn poll(self: Pin<&mut Self>, cx: &mut TContext<'_>) -> Poll<Self::Output> {
         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<F: Future> Future for WarnAfterFuture<F> {
                     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
                     );
                 }