diff --git a/backend/src/main.rs b/backend/src/main.rs index dcf739215f..d1890ebf3c 100644 --- a/backend/src/main.rs +++ b/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(()), } }; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 0fb0bde738..4bd08f407c 100644 --- a/backend/windmill-common/src/lib.rs +++ b/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 = 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"); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 124e3880cd..c1ff9c3542 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/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>) -> 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);