feat(backend): add /ready endpoint for workers

backend/src/main.rs

@@ -156,9 +156,11 @@ Windmill Community Edition {GIT_VERSION}
 
         let metrics_f = async {
             match metrics_addr {
-                Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe())
-                    .await
-                    .map_err(anyhow::Error::from),
+                Some(addr) => {
+                    windmill_common::serve_metrics(addr, rx.resubscribe(), num_workers > 0)
+                        .await
+                        .map_err(anyhow::Error::from)
+                }
                 None => Ok(()),
             }
         };

backend/windmill-common/src/lib.rs

@@ -6,7 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-use std::net::SocketAddr;
+use std::{net::SocketAddr, sync::Arc};
 
 use error::Error;
 
@@ -31,6 +31,7 @@ pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 3;
 
 lazy_static::lazy_static! {
     pub static ref BASE_URL: String = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string());
+    pub static ref IS_READY: Arc<std::sync::atomic::AtomicBool> = Arc::new(std::sync::atomic::AtomicBool::new(false));
 }
 
 #[cfg(feature = "tokio")]
@@ -58,14 +59,31 @@ pub async fn shutdown_signal(tx: tokio::sync::broadcast::Sender<()>) -> anyhow::
 pub async fn serve_metrics(
     addr: SocketAddr,
     mut rx: tokio::sync::broadcast::Receiver<()>,
+    ready_worker_endpoint: bool,
 ) -> Result<(), hyper::Error> {
+    use std::sync::atomic::Ordering;
+
     use axum::{routing::get, Router};
-    axum::Server::bind(&addr)
-        .serve(
-            Router::new()
-                .route("/metrics", get(metrics))
-                .into_make_service(),
+    use hyper::StatusCode;
+    let router = Router::new().route("/metrics", get(metrics));
+
+    let router = if ready_worker_endpoint {
+        router.route(
+            "/ready",
+            get(|| async {
+                if IS_READY.load(Ordering::Relaxed) {
+                    (StatusCode::OK, "ready")
+                } else {
+                    (StatusCode::INTERNAL_SERVER_ERROR, "not ready")
+                }
+            }),
         )
+    } else {
+        router
+    };
+
+    axum::Server::bind(&addr)
+        .serve(router.into_make_service())
         .with_graceful_shutdown(async {
             rx.recv().await.ok();
             println!("Graceful shutdown of metrics");

backend/windmill-worker/src/worker.rs

@@ -16,7 +16,7 @@ use sqlx::{Pool, Postgres, Transaction};
 use windmill_api_client::Client;
 use std::{
     borrow::Borrow, collections::HashMap, io, os::unix::process::ExitStatusExt, panic,
-    process::Stdio, time::Duration,
+    process::Stdio, time::Duration, sync::atomic::Ordering,
 };
 use tracing::{trace_span, Instrument};
 use uuid::Uuid;
@@ -26,7 +26,7 @@ use windmill_common::{
     flows::{FlowModuleValue, FlowValue},
     scripts::{ScriptHash, ScriptLang},
     utils::rd_string,
-    variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL,
+    variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY,
 };
 use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob, CLOUD_HOSTED};
 
@@ -74,6 +74,7 @@ async fn copy_cache_from_bucket(bucket: &str, tx: Option<Sender<()>>) -> Option:
             .arg("copy")
             .arg(format!(":s3,env_auth=true:{bucket}"))
             .arg(if tx_is_some { ROOT_TMP_CACHE_DIR } else { ROOT_CACHE_DIR })
+            .arg("-vv")
             .arg("--size-only")
             .arg("--fast-list")
             .arg("--exclude")
@@ -123,6 +124,7 @@ async fn copy_cache_to_bucket(bucket: &str) {
         .arg("copy")
         .arg(ROOT_CACHE_DIR)
         .arg(format!(":s3,env_auth=true:{bucket}"))
+        .arg("-vv")
         .arg("--size-only")
         .arg("--fast-list")
         .arg("--exclude")
@@ -181,6 +183,7 @@ async fn copy_cache_to_bucket_as_tar(bucket: &str) {
         .arg("copyto")
         .arg(TAR_CACHE_FILENAME)
         .arg(format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"))
+        .arg("-vv")
         .arg("--size-only")
         .arg("--fast-list")
         .stdin(Stdio::null())
@@ -208,6 +211,7 @@ async fn copy_cache_from_bucket_as_tar(bucket: &str) -> bool {
         .arg("copyto")
         .arg(format!(":s3,env_auth=true:{bucket}/{TAR_CACHE_FILENAME}"))
         .arg(format!("{ROOT_CACHE_DIR}{TAR_CACHE_FILENAME}"))
+        .arg("-vv")
         .arg("--size-only")
         .arg("--fast-list")
         .stdin(Stdio::null())
@@ -603,6 +607,8 @@ pub async fn run_worker(
         }
     }
 
+    IS_READY.store(true, Ordering::Relaxed);
+
     tracing::info!(worker = %worker_name, "starting worker");
 
     #[cfg(feature = "enterprise")]
@@ -613,6 +619,7 @@ pub async fn run_worker(
 
     tracing::info!(worker = %worker_name, "listening for jobs");
 
+    
     loop {
         worker_busy.set(0);