diff --git a/backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json b/backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json new file mode 100644 index 0000000000..9304f1c75d --- /dev/null +++ b/backend/.sqlx/query-ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "wm_version", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "ad03e5acf10ef94abc37cb9f56b1775c67f075c8bc83458e8be2e242347218d6" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d37f3cb23a..f239021c99 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -10811,6 +10811,7 @@ dependencies = [ "rand 0.8.5", "regex", "reqwest 0.12.9", + "semver 1.0.23", "serde", "serde_json", "sha2 0.10.8", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 800e165e33..51bf578c96 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -180,6 +180,7 @@ tokio-util = { version = "^0", features = ["io"] } json-pointer = "^0" itertools = "^0" regex = "^1" +semver = "^1" deno_fetch = "0.195.0" deno_tls = "0.158.0" diff --git a/backend/src/main.rs b/backend/src/main.rs index 6869a24647..1f1fab319d 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -7,7 +7,6 @@ */ use anyhow::Context; -use git_version::git_version; use monitor::{ reload_timeout_wait_result_setting, send_current_log_file_to_object_store, send_logs_to_object_store, @@ -42,7 +41,7 @@ use windmill_common::{ }, scripts::ScriptLang, stats_ee::schedule_stats, - utils::{hostname, rd_string, Mode}, + utils::{hostname, rd_string, Mode, GIT_VERSION}, worker::{reload_custom_tags_setting, HUB_CACHE_DIR, TMP_DIR, WORKER_GROUP}, DB, METRICS_ENABLED, }; @@ -84,7 +83,6 @@ use crate::monitor::{ #[cfg(feature = "parquet")] use crate::monitor::reload_s3_cache_setting; -const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); const DEFAULT_NUM_WORKERS: usize = 1; const DEFAULT_PORT: u16 = 8000; const DEFAULT_SERVER_BIND_ADDR: Ipv4Addr = Ipv4Addr::new(0, 0, 0, 0); diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index da811bb6ff..b40c89c766 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -51,8 +51,8 @@ use windmill_common::{ utils::{now_from_db, rd_string, report_critical_error, Mode}, worker::{ load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting, - DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG, WORKER_CONFIG, - WORKER_GROUP, + update_min_version, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SMTP_CONFIG, + WORKER_CONFIG, WORKER_GROUP, }, BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED, CRITICAL_ALERT_MUTE_UI_ENABLED @@ -1080,6 +1080,10 @@ pub async fn monitor_db( } }; + let update_min_worker_version_f = async { + update_min_version(db).await; + }; + join!( expired_items_f, zombie_jobs_f, @@ -1088,6 +1092,7 @@ pub async fn monitor_db( worker_groups_alerts_f, jobs_waiting_alerts_f, apply_autoscaling_f, + update_min_worker_version_f, ); } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 5a7f7fcded..fe74584d1b 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -25,7 +25,6 @@ use argon2::Argon2; use axum::extract::DefaultBodyLimit; use axum::{middleware::from_extractor, routing::get, Extension, Router}; use db::DB; -use git_version::git_version; use http::HeaderValue; use reqwest::Client; use std::collections::HashMap; @@ -40,7 +39,7 @@ use tower_http::{ }; use windmill_common::db::UserDB; use windmill_common::worker::{ALL_TAGS, CLOUD_HOSTED}; -use windmill_common::{BASE_URL, INSTANCE_NAME}; +use windmill_common::{BASE_URL, INSTANCE_NAME, utils::GIT_VERSION}; use crate::scim_ee::has_scim_token; use windmill_common::error::AppError; @@ -93,9 +92,6 @@ mod workers; mod workspaces; mod workspaces_ee; -pub const GIT_VERSION: &str = - git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); - pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB lazy_static::lazy_static! { diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index f3adbfde84..fd0f3aa8c1 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -58,6 +58,7 @@ async-stream.workspace = true const_format.workspace = true crc.workspace = true windmill-macros.workspace = true +semver.workspace = true [target.'cfg(not(target_env = "msvc"))'.dependencies] tikv-jemalloc-ctl = { optional = true, workspace = true } diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 18c0c93205..3223ffb75f 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -21,6 +21,7 @@ use reqwest::Client; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use sqlx::{Pool, Postgres}; +use semver::Version; pub const MAX_PER_PAGE: usize = 10000; pub const DEFAULT_PER_PAGE: usize = 1000; @@ -37,6 +38,10 @@ lazy_static::lazy_static! { .timeout(std::time::Duration::from_secs(20)) .connect_timeout(std::time::Duration::from_secs(10)) .build().unwrap(); + pub static ref GIT_SEM_VERSION: Version = Version::parse( + // skip first `v` character. + GIT_VERSION.split_at(1).1 + ).unwrap_or(Version::new(0, 1, 0)); } #[derive(Deserialize, Clone)] diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 37ae61c8d6..eb017bc79f 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1,6 +1,7 @@ use const_format::concatcp; use itertools::Itertools; use regex::Regex; +use semver::Version; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use std::{ @@ -88,6 +89,7 @@ lazy_static::lazy_static! { .and_then(|x| x.parse::().ok()) .unwrap_or(false); + pub static ref MIN_VERSION: Arc> = Arc::new(RwLock::new(Version::new(0, 0, 0))); } pub async fn make_suspended_pull_query(wc: &WorkerConfig) { @@ -548,6 +550,31 @@ pub fn get_windmill_memory_usage() -> Option { } } +pub async fn update_min_version<'c, E: sqlx::Executor<'c, Database = sqlx::Postgres>>(executor: E) -> bool { + use crate::utils::{GIT_VERSION, GIT_SEM_VERSION}; + + // fetch all pings with a different version than self from the last 5 minutes. + let pings = sqlx::query_scalar!( + "SELECT wm_version FROM worker_ping WHERE wm_version != $1 AND ping_at > now() - interval '5 minutes'", + GIT_VERSION + ).fetch_all(executor).await.unwrap_or_default(); + + let cur_version = GIT_SEM_VERSION.clone(); + let min_version = pings + .iter() + .filter(|x| !x.is_empty()) + .filter_map(|x| semver::Version::parse(x.split_at(1).1).ok()) + .min() + .unwrap_or_else(|| cur_version.clone()); + + if min_version != cur_version { + tracing::info!("Minimal worker version: {min_version}"); + } + + *MIN_VERSION.write().await = min_version.clone(); + min_version >= cur_version +} + pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) { let (tags, dw) = { let wc = WORKER_CONFIG.read().await.clone();