starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: worker groups admin panel (#2277)

Browse Source 
* merge

* merge

* merge

* wg

* progress

* all

* all

* all

* all

* all

* all

* fix

* fix
This commit is contained in:
Ruben Fiszel
2023-09-13 21:24:54 +02:00
committed by GitHub
parent e85ee6cf1e
commit dfcdb2d8ff
37 changed files with 1273 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-03c7f098ad795d216d58ded0bf4cf6473960377455b5fd7ac3b578a1d36c0cc6.json
Generated
+19
View File
@@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"TextArray",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "03c7f098ad795d216d58ded0bf4cf6473960377455b5fd7ac3b578a1d36c0cc6"
}
backend/.sqlx/query-210b7fa50246d9b8fd1a24ade787bad6882c86e9422b715844aa92b67ed05174.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM worker_group_config WHERE name = $1 RETURNING name",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "name",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false
]
},
"hash": "210b7fa50246d9b8fd1a24ade787bad6882c86e9422b715844aa92b67ed05174"
}
backend/.sqlx/query-4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47.json → backend/.sqlx/query-240ce8c9b5c7530999642190c6f7915ae2734b90b8c4cd35fe37783b1d4dd0b0.json
Generated
+9 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2",
"describe": {
"columns": [
{
@@ -37,6 +37,11 @@
"ordinal": 6,
"name": "custom_tags",
"type_info": "TextArray"
},
{
"ordinal": 7,
"name": "worker_group",
"type_info": "Varchar"
}
],
"parameters": {
@@ -52,8 +57,9 @@
false,
false,
false,
true
true,
false
]
},
"hash": "4f6b3b472b4b78c0325cf3755f9ef1806d2e82328ceccbeade8cc2333c6dfe47"
"hash": "240ce8c9b5c7530999642190c6f7915ae2734b90b8c4cd35fe37783b1d4dd0b0"
}
backend/.sqlx/query-07551a32c49da8c0693dd39c6a63b5b2a596ccc0e52e8918160604a5e133dd32.json → backend/.sqlx/query-47beea5cd6324b53bfb349665fb215280f32b70a617fde87f70ea53ca9ade39f.json
Generated
+3 -2
View File
@@ -1,15 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1 WHERE worker = $2",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Int4",
"TextArray",
"Text"
]
},
"nullable": []
},
"hash": "07551a32c49da8c0693dd39c6a63b5b2a596ccc0e52e8918160604a5e133dd32"
"hash": "47beea5cd6324b53bfb349665fb215280f32b70a617fde87f70ea53ca9ade39f"
}
backend/.sqlx/query-61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf.json
Generated
-17
View File
@@ -1,17 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags) VALUES ($1, $2, $3, $4) ON CONFLICT (worker) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"TextArray"
]
},
"nullable": []
},
"hash": "61e6aac871b482b6e36f866b4ec9148a75e1bd130e7614463487e2ba6957dfdf"
}
backend/.sqlx/query-6c0136f7965f1e01620a7d5efd7edbc78f0bf3f815676d8b343152bfce2dfc4c.json
Generated
+22
View File
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "SELECT config FROM worker_group_config WHERE name = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "config",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
true
]
},
"hash": "6c0136f7965f1e01620a7d5efd7edbc78f0bf3f815676d8b343152bfce2dfc4c"
}
backend/.sqlx/query-7998b23eb72f5967a0fd376fa1015bf29ba5150cd59732288ceb72ce9cecf987.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "PostgreSQL",
"query": "SELECT * FROM worker_group_config",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "name",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "config",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
true
]
},
"hash": "7998b23eb72f5967a0fd376fa1015bf29ba5150cd59732288ceb72ce9cecf987"
}
backend/.sqlx/query-903f2f62f3829274f5dfa0cb1ebbb9e132d310d19ca744531a42ca2c7ac56f56.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO worker_group_config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Jsonb"
]
},
"nullable": []
},
"hash": "903f2f62f3829274f5dfa0cb1ebbb9e132d310d19ca744531a42ca2c7ac56f56"
}
backend/Cargo.lock
Generated
+2
View File
@@ -7212,9 +7212,11 @@ dependencies = [
"hex",
"hmac",
"hyper",
"itertools 0.10.5",
"lazy_static",
"prometheus",
"rand 0.8.5",
"regex",
"reqwest",
"serde",
"serde_json",
backend/migrations/20230908155455_add_worker_group.down.sql
+4
View File
@@ -0,0 +1,4 @@
-- Add down migration script here
DROP TABLE worker_group_config;
ALTER TABLE worker_ping DROP COLUMN worker_group;
backend/migrations/20230908155455_add_worker_group.up.sql
+8
View File
@@ -0,0 +1,8 @@
-- Add up migration script here
CREATE TABLE worker_group_config (
name VARCHAR(255) PRIMARY KEY,
config JSONB DEFAULT '{}'::jsonb
);
ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS worker_group VARCHAR(255) NOT NULL DEFAULT 'default';
ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS dedicated_worker VARCHAR(255);
backend/src/main.rs
+60 -39
View File
@@ -8,15 +8,14 @@
use gethostname::gethostname;
use git_version::git_version;
use monitor::handle_zombie_jobs_periodically;
use sqlx::{Pool, Postgres};
use std::{
net::{IpAddr, Ipv4Addr, SocketAddr},
sync::Arc,
time::Duration,
};
use tokio::{
fs::{metadata, DirBuilder},
join,
sync::RwLock,
};
use windmill_api::{LICENSE_KEY, OAUTH_CLIENTS, SMTP_CLIENT};
@@ -28,6 +27,8 @@ use windmill_worker::{
PIP_CACHE_DIR, ROOT_TMP_CACHE_DIR, TAR_PIP_TMP_CACHE_DIR,
};
use crate::monitor::monitor_db;
const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
const DEFAULT_NUM_WORKERS: usize = 1;
const DEFAULT_PORT: u16 = 8000;
@@ -135,7 +136,10 @@ Windmill Community Edition {GIT_VERSION}
tracing::info!("Smtp client connected.");
}
}
if server_mode || num_workers > 0 {
let worker_mode = num_workers > 0;
if server_mode || worker_mode {
let port_var = std::env::var("PORT").ok().and_then(|x| x.parse().ok());
let port = if server_mode {
@@ -144,6 +148,19 @@ Windmill Community Edition {GIT_VERSION}
port_var.unwrap_or(0)
};
// since it's only on server mode, the port is statically defined
let base_internal_url: String = format!("http://localhost:{}", port.to_string());
monitor_db(
&db,
tx.clone(),
&base_internal_url,
rsmq.clone(),
worker_mode,
server_mode,
)
.await;
if std::env::var("BASE_INTERNAL_URL").is_ok() {
tracing::warn!("BASE_INTERNAL_URL is now unecessary and ignored, you can remove it.");
}
@@ -161,10 +178,11 @@ Windmill Community Edition {GIT_VERSION}
let workers_f = async {
let port = port_rx.await?;
let base_internal_url: String = format!("http://localhost:{}", port.to_string());
if num_workers > 0 {
if worker_mode {
run_workers(
db.clone(),
rx.resubscribe(),
tx.clone(),
num_workers,
base_internal_url.clone(),
rsmq.clone(),
@@ -176,29 +194,48 @@ Windmill Community Edition {GIT_VERSION}
Ok(()) as anyhow::Result<()>
};
let rsmq2 = rsmq.clone();
let monitor_f = async {
if server_mode {
// since it's only on server mode, the port is statically defined
let base_internal_url: String = format!("http://localhost:{}", port.to_string());
monitor_db(&db, rx.resubscribe(), &base_internal_url, rsmq2).await;
}
let db = db.clone();
let tx = tx.clone();
let rsmq = rsmq.clone();
let mut rx = rx.resubscribe();
let base_internal_url = base_internal_url.to_string();
tokio::spawn(async move {
//monitor_db is applied at start, no need to apply it twice
tokio::time::sleep(Duration::from_secs(30)).await;
loop {
monitor_db(
&db,
tx.clone(),
&base_internal_url,
rsmq.clone(),
worker_mode,
server_mode,
)
.await;
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(30)) => (),
_ = rx.recv() => {
println!("received killpill for monitor job");
break;
}
}
}
});
Ok(()) as anyhow::Result<()>
};
let metrics_f = async {
match metrics_addr {
Some(_addr) => {
#[cfg(not(feature = "enterprise"))]
panic!("Metrics are only available in the Enterprise Edition");
if let Some(_addr) = metrics_addr {
#[cfg(not(feature = "enterprise"))]
panic!("Metrics are only available in the Enterprise Edition");
#[cfg(feature = "enterprise")]
windmill_common::serve_metrics(_addr, rx.resubscribe(), num_workers > 0)
.await
.map_err(anyhow::Error::from)
}
None => Ok(()),
#[cfg(feature = "enterprise")]
windmill_common::serve_metrics(_addr, rx.resubscribe(), num_workers > 0).await;
}
Ok(()) as anyhow::Result<()>
};
futures::try_join!(shutdown_signal, server_f, metrics_f, workers_f, monitor_f)?;
@@ -225,28 +262,10 @@ fn display_config(envs: &[&str]) {
)
}
pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: &Pool<Postgres>,
rx: tokio::sync::broadcast::Receiver<()>,
base_internal_url: &str,
rsmq: Option<R>,
) -> tokio::task::JoinHandle<()> {
let db1 = db.clone();
let db2 = db.clone();
let rx2 = rx.resubscribe();
let base_internal_url = base_internal_url.to_string();
tokio::spawn(async move {
join!(
handle_zombie_jobs_periodically(&db1, rx, &base_internal_url, rsmq),
windmill_api::delete_expired_items_perdiodically(&db2, rx2)
);
})
}
pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: Pool<Postgres>,
rx: tokio::sync::broadcast::Receiver<()>,
tx: tokio::sync::broadcast::Sender<()>,
num_workers: i32,
base_internal_url: String,
rsmq: Option<R>,
@@ -320,6 +339,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
let worker_name = format!("wk-{}-{}", &instance_name, rd_string(5));
let ip = ip.clone();
let rx = rx.resubscribe();
let tx = tx.clone();
let base_internal_url = base_internal_url.clone();
let rsmq2 = rsmq.clone();
let sync_barrier = sync_barrier.clone();
@@ -333,6 +353,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
num_workers as u32,
&ip,
rx,
tx,
&base_internal_url,
rsmq2,
sync_barrier,
backend/src/monitor.rs
+52 -15
View File
@@ -1,12 +1,11 @@
use std::time::Duration;
use once_cell::sync::OnceCell;
use sqlx::{Pool, Postgres};
use tokio::sync::mpsc;
use tokio::{join, sync::mpsc};
use uuid::Uuid;
use windmill_common::{
error,
jobs::{JobKind, QueuedJob},
worker::{load_worker_config, reload_custom_tags_setting, WORKER_CONFIG},
METRICS_ENABLED,
};
use windmill_worker::{
@@ -37,24 +36,62 @@ lazy_static::lazy_static! {
.unwrap();
}
pub async fn handle_zombie_jobs_periodically<
R: rsmq_async::RsmqConnection + Send + Sync + Clone,
>(
pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: &Pool<Postgres>,
mut rx: tokio::sync::broadcast::Receiver<()>,
tx: tokio::sync::broadcast::Sender<()>,
base_internal_url: &str,
rsmq: Option<R>,
worker_mode: bool,
server_mode: bool,
) {
loop {
handle_zombie_jobs(db, base_internal_url, rsmq.clone()).await;
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(30)) => (),
_ = rx.recv() => {
println!("received killpill for monitor job");
break;
let zombie_jobs_f = async {
if server_mode {
handle_zombie_jobs(db, base_internal_url, rsmq.clone()).await;
}
};
let expired_items_f = async {
if server_mode {
windmill_api::delete_expired_items(&db).await;
}
};
let reload_worker_config_f = async {
if worker_mode {
reload_worker_config(&db, tx).await;
}
};
let reload_custom_tags_f = async {
if server_mode {
if let Err(e) = reload_custom_tags_setting(db).await {
tracing::error!("Error reloading custom tags: {:?}", e)
}
}
};
join!(
expired_items_f,
zombie_jobs_f,
reload_worker_config_f,
reload_custom_tags_f
);
}
pub async fn reload_worker_config(db: &Pool<Postgres>, tx: tokio::sync::broadcast::Sender<()>) {
let config = load_worker_config(&db).await;
if let Err(e) = config {
tracing::error!("Error reloading worker config: {:?}", e)
} else {
let wc = WORKER_CONFIG.read().await;
let config = config.unwrap();
if *wc != config {
if (*wc).dedicated_worker != config.dedicated_worker {
tracing::info!("Dedicated worker config changed, sending killpill. Expecting to be restarted by supervisor.");
let _ = tx.send(());
}
drop(wc);
let mut wc = WORKER_CONFIG.write().await;
tracing::info!("Reloading worker config...");
*wc = config
}
}
}
backend/tests/worker.rs
+2
View File
@@ -956,6 +956,7 @@ fn spawn_test_worker(
let worker_name: String = next_worker_name();
let ip: &str = Default::default();
let tx2 = tx.clone();
let future = async move {
let base_internal_url = format!("http://localhost:{}", port);
windmill_worker::run_worker::<rsmq_async::MultiplexedRsmq>(
@@ -966,6 +967,7 @@ fn spawn_test_worker(
1,
ip,
rx,
tx2,
&base_internal_url,
None,
Arc::new(RwLock::new(None)),
backend/windmill-api/openapi.yaml
+62
View File
@@ -5252,6 +5252,65 @@ paths:
items:
$ref: "#/components/schemas/WorkerPing"
/workers/list_worker_groups:
get:
summary: list workers
operationId: listWorkerGroups
tags:
- worker
responses:
"200":
description: a list of workers
content:
application/json:
schema:
type: array
items:
type: object
properties:
name:
type: string
config: {}
required:
- name
- config
/workers/worker_group/{name}:
post:
summary: Update Worker Group
operationId: updateWorkerGroup
tags:
- worker
parameters:
- $ref: "#/components/parameters/Name"
requestBody:
description: worker group
required: true
content:
application/json:
schema: {}
responses:
"200":
description: Update a worker group
content:
text/plain:
schema:
type: string
delete:
summary: Delete Worker Group
operationId: deleteWorkerGroup
tags:
- worker
parameters:
- $ref: "#/components/parameters/Name"
responses:
"200":
description: Delete a worker group
content:
text/plain:
schema:
type: string
/w/{workspace}/acls/get/{kind}/{path}:
get:
summary: get granular acls
@@ -7172,6 +7231,8 @@ components:
type: array
items:
type: string
worker_group:
type: string
required:
- worker
- worker_instance
@@ -7179,6 +7240,7 @@ components:
- started_at
- ip
- jobs_executed
- worker_group
UserWorkspaceList:
type: object
backend/windmill-api/src/groups.rs
+1 -1
View File
@@ -15,6 +15,7 @@ use axum::{
Json, Router,
};
use windmill_audit::{audit_log, ActionKind};
use windmill_common::worker::CLOUD_HOSTED;
use windmill_common::{db::UserDB, users::username_to_permissioned_as};
use windmill_common::{
error::{Error, JsonResult, Result},
@@ -23,7 +24,6 @@ use windmill_common::{
use serde::{Deserialize, Serialize};
use sqlx::{query_scalar, FromRow, Postgres, Transaction};
use windmill_queue::CLOUD_HOSTED;
pub fn workspaced_service() -> Router {
Router::new()
backend/windmill-api/src/jobs.rs
+13 -13
View File
@@ -13,7 +13,6 @@ use crate::{
users::{check_scopes, require_owner_of_path, OptAuthed},
utils::require_super_admin,
variables::get_workspace_key,
workers::{CUSTOM_TAGS, CUSTOM_TAGS_PER_WORKSPACE},
BASE_URL,
};
use anyhow::Context;
@@ -34,6 +33,7 @@ use sqlx::{query_scalar, types::Uuid, FromRow, Postgres, Transaction};
use tower_http::cors::{Any, CorsLayer};
use urlencoding::encode;
use windmill_audit::{audit_log, ActionKind};
use windmill_common::worker::CUSTOM_TAGS_PER_WORKSPACE;
use windmill_common::{
db::UserDB,
error::{self, to_anyhow, Error},
@@ -1607,12 +1607,12 @@ fn add_raw_string(
return args;
}
fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error::Result<()> {
async fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error::Result<()> {
if let Some(tag) = tag {
if tag == "" {
return Ok(());
}
let custom_tags_per_w = &*CUSTOM_TAGS_PER_WORKSPACE;
let custom_tags_per_w = CUSTOM_TAGS_PER_WORKSPACE.read().await;
if custom_tags_per_w.0.contains(&tag.to_string()) {
Ok(())
} else if custom_tags_per_w.1.contains_key(tag)
@@ -1626,7 +1626,7 @@ fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error:
} else {
return Err(error::Error::BadRequest(format!(
"Tag {tag} cannot be used on workspace {w_id}: (CUSTOM_TAGS: {:?})",
*CUSTOM_TAGS
custom_tags_per_w
)));
}
} else {
@@ -1655,7 +1655,7 @@ pub async fn run_flow_by_path(
.fetch_optional(&db)
.await?
.flatten();
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
let args = add_raw_string(raw_string, args);
@@ -1705,7 +1705,7 @@ pub async fn run_job_by_path(
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
let args = add_raw_string(raw_string, args);
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -1908,7 +1908,7 @@ pub async fn run_wait_result_job_by_path_get(
check_scopes(&authed, || format!("run:script/{script_path}"))?;
let (job_payload, tag) = script_path_to_payload(script_path, &db, &w_id).await?;
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2079,7 +2079,7 @@ async fn run_wait_result_script_by_path_internal(
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
let args = add_raw_string(raw_string, args);
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2143,7 +2143,7 @@ pub async fn run_wait_result_script_by_hash(
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
let args = add_raw_string(raw_string, args);
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2258,7 +2258,7 @@ async fn run_wait_result_flow_by_path_internal(
.fetch_optional(&db)
.await?
.flatten();
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2314,7 +2314,7 @@ async fn run_preview_job(
}
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default());
check_tag_available_for_workspace(&w_id, &preview.tag)?;
check_tag_available_for_workspace(&w_id, &preview.tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2422,7 +2422,7 @@ async fn run_preview_flow_job(
}
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default());
check_tag_available_for_workspace(&w_id, &raw_flow.tag)?;
check_tag_available_for_workspace(&w_id, &raw_flow.tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -2479,7 +2479,7 @@ pub async fn run_job_by_hash(
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let args = run_query.add_include_headers(headers, args.unwrap_or_default());
let args = add_raw_string(raw_string, args);
check_tag_available_for_workspace(&w_id, &tag)?;
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into(), rsmq);
let (uuid, tx) = push(
backend/windmill-api/src/lib.rs
+3 -3
View File
@@ -11,7 +11,6 @@ use crate::oauth2::AllClients;
use crate::saml::{SamlSsoLogin, ServiceProviderExt};
use crate::scim::has_scim_token;
use crate::tracing_init::MyOnFailure;
use crate::workers::ALL_TAGS;
use crate::{
oauth2::{build_oauth_clients, SlackVerifier},
tracing_init::{MyMakeSpan, MyOnResponse},
@@ -36,6 +35,7 @@ use tower_http::{
};
use windmill_common::db::UserDB;
use windmill_common::utils::rd_string;
use windmill_common::worker::ALL_TAGS;
use windmill_common::error::AppError;
@@ -72,7 +72,7 @@ mod workspaces;
pub const GIT_VERSION: &str =
git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
pub use users::delete_expired_items_perdiodically;
pub use users::delete_expired_items;
pub const DEFAULT_BODY_LIMIT: usize = 2097152; // 2MB
lazy_static::lazy_static! {
@@ -153,7 +153,7 @@ pub async fn run_server(
port_tx: tokio::sync::oneshot::Sender<u16>,
) -> anyhow::Result<()> {
if let Some(mut rsmq) = rsmq.clone() {
for tag in ALL_TAGS.clone() {
for tag in ALL_TAGS.read().await.iter() {
let r =
rsmq_async::RsmqConnection::create_queue(&mut rsmq, &tag, None, None, None).await;
if let Err(e) = r {
backend/windmill-api/src/workers.rs
+92 -49
View File
@@ -7,74 +7,44 @@
*/
use axum::{
extract::{Extension, Query},
extract::{Extension, Path, Query},
routing::get,
Json, Router,
};
use itertools::Itertools;
use regex::Regex;
use serde::{Deserialize, Serialize};
use sqlx::FromRow;
use windmill_common::{
db::UserDB,
error::JsonResult,
error::{self, JsonResult},
utils::{paginate, Pagination},
worker::ALL_TAGS,
DB,
};
use std::collections::HashMap;
#[cfg(feature = "benchmark")]
use std::sync::atomic::Ordering;
#[cfg(feature = "benchmark")]
use windmill_queue::IDLE_WORKERS;
use crate::db::ApiAuthed;
use crate::{db::ApiAuthed, utils::require_super_admin};
#[cfg(not(feature = "benchmark"))]
pub fn global_service() -> Router {
Router::new()
use axum::routing::post;
let router = Router::new()
.route("/list", get(list_worker_pings))
.route("/custom_tags", get(get_custom_tags))
}
.route("/list_worker_groups", get(get_worker_groups))
.route(
"/worker_group/:name",
post(update_worker_group).delete(delete_worker_group),
);
#[cfg(feature = "benchmark")]
return router.route("/toggle", get(toggle));
#[cfg(feature = "benchmark")]
pub fn global_service() -> Router {
Router::new()
.route("/toggle", get(toggle))
.route("/list", get(list_worker_pings))
.route("/custom_tags", get(get_custom_tags))
}
lazy_static::lazy_static! {
pub static ref CUSTOM_TAGS: Vec<String> = std::env::var("CUSTOM_TAGS")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect::<Vec<_>>()).unwrap_or_default();
pub static ref CUSTOM_TAGS_PER_WORKSPACE: (Vec<String>, HashMap<String, Vec<String>>) = process_custom_tags(std::env::var("CUSTOM_TAGS")
.ok());
pub static ref ALL_TAGS: Vec<String> = [CUSTOM_TAGS_PER_WORKSPACE.0.clone(), CUSTOM_TAGS_PER_WORKSPACE.1.keys().map(|x| x.to_string()).collect_vec()].concat();
}
fn process_custom_tags(o: Option<String>) -> (Vec<String>, HashMap<String, Vec<String>>) {
let regex = Regex::new(r"^(\w+)\(((?:\w+)\+?)+\)$").unwrap();
if let Some(s) = o {
let mut global = vec![];
let mut specific: HashMap<String, Vec<String>> = HashMap::new();
for e in s.split(",") {
if let Some(cap) = regex.captures(e) {
let tag = cap.get(1).unwrap().as_str().to_string();
let workspaces = cap.get(2).unwrap().as_str().split("+");
specific.insert(tag, workspaces.map(|x| x.to_string()).collect_vec());
} else {
global.push(e.to_string());
}
}
(global, specific)
} else {
(vec![], HashMap::new())
}
#[cfg(not(feature = "benchmark"))]
return router;
}
#[derive(FromRow, Serialize, Deserialize)]
@@ -86,6 +56,7 @@ struct WorkerPing {
ip: String,
jobs_executed: i32,
custom_tags: Option<Vec<String>>,
worker_group: String,
}
#[derive(Serialize, Deserialize)]
@@ -104,7 +75,7 @@ async fn list_worker_pings(
let rows = sqlx::query_as!(
WorkerPing,
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2",
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, custom_tags, worker_group FROM worker_ping ORDER BY ping_at desc LIMIT $1 OFFSET $2",
per_page as i64,
offset as i64
)
@@ -121,5 +92,77 @@ async fn toggle(Query(query): Query<EnableWorkerQuery>) -> JsonResult<bool> {
}
async fn get_custom_tags() -> Json<Vec<String>> {
Json(ALL_TAGS.clone())
Json(ALL_TAGS.read().await.clone().into())
}
#[derive(Serialize, Deserialize, FromRow)]
struct WorkerGroup {
name: String,
config: serde_json::Value,
}
async fn get_worker_groups(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
) -> error::JsonResult<Vec<WorkerGroup>> {
let mut tx = user_db.begin(&authed).await?;
require_super_admin(&db, &authed.email).await?;
let rows = sqlx::query_as!(WorkerGroup, "SELECT * FROM worker_group_config")
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
Ok(Json(rows))
}
async fn update_worker_group(
Path(name): Path<String>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
authed: ApiAuthed,
Json(config): Json<serde_json::Value>,
) -> error::Result<String> {
let tx = user_db.begin(&authed).await?;
require_super_admin(&db, &authed.email).await?;
tx.commit().await?;
sqlx::query!(
"INSERT INTO worker_group_config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = $2",
&name,
config
)
.execute(&db)
.await?;
Ok(format!("Updated worker group {name}"))
}
async fn delete_worker_group(
Path(name): Path<String>,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
authed: ApiAuthed,
) -> error::Result<String> {
let tx = user_db.begin(&authed).await?;
require_super_admin(&db, &authed.email).await?;
tx.commit().await?;
let deleted = sqlx::query!(
"DELETE FROM worker_group_config WHERE name = $1 RETURNING name",
name,
)
.fetch_all(&db)
.await?;
if deleted.len() == 0 {
return Err(error::Error::NotFound(format!(
"Worker group {name} not found",
name = name
)));
}
Ok(format!("Deleted worker group {name}"))
}
backend/windmill-api/src/workspaces.rs
+12 -6
View File
@@ -74,13 +74,19 @@ pub fn workspaced_service() -> Router {
.route("/edit_error_handler", post(edit_error_handler));
#[cfg(feature = "enterprise")]
tracing::info!("stripe enabled");
#[cfg(feature = "enterprise")]
let router = router
.route("/checkout", get(stripe_checkout))
.route("/billing_portal", get(stripe_portal));
{
if std::env::var("STRIPE_KEY").is_err() {
return router;
} else {
tracing::info!("stripe enabled");
return router
.route("/checkout", get(stripe_checkout))
.route("/billing_portal", get(stripe_portal));
}
}
#[cfg(not(feature = "enterprise"))]
router
}
pub fn global_service() -> Router {
backend/windmill-common/Cargo.toml
+2
View File
@@ -44,3 +44,5 @@ reqwest = { workspace = true, optional = true }
tracing-subscriber = { workspace = true, optional = true }
lazy_static.workspace = true
tracing-flame = { version = "^0", optional = true }
itertools.workspace = true
regex.workspace = true
backend/windmill-common/src/global_settings.rs
+3 -3
View File
@@ -1,6 +1,7 @@
pub const WORKER_S3_BUCKET_SYNC: &str = "worker_s3_bucket_sync";
pub const CUSTOM_TAGS_SETTING: &str = "custom_tags";
pub const ENV_SETTINGS: [&str; 55] = [
pub const ENV_SETTINGS: [&str; 54] = [
"DISABLE_NSJAIL",
"DISABLE_SERVER",
"NUM_WORKERS",
@@ -41,8 +42,6 @@ pub const ENV_SETTINGS: [&str; 55] = [
"INSTANCE_EVENTS_WEBHOOK",
"CLOUD_HOSTED",
"GLOBAL_CACHE_INTERVAL",
"WORKER_TAGS",
"CUSTOM_TAGS",
"JOB_RETENTION_SECS",
"WAIT_RESULT_FAST_POLL_DURATION_SECS",
"WAIT_RESULT_SLOW_POLL_INTERVAL_MS",
@@ -56,4 +55,5 @@ pub const ENV_SETTINGS: [&str; 55] = [
"CREATE_WORKSPACE_REQUIRE_SUPERADMIN",
"GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE",
"MAX_WAIT_FOR_SIGTERM",
"WORKER_GROUP",
];
backend/windmill-common/src/lib.rs
+17 -8
View File
@@ -27,6 +27,7 @@ pub mod scripts;
pub mod users;
pub mod utils;
pub mod variables;
pub mod worker;
#[cfg(feature = "tracing_init")]
pub mod tracing_init;
@@ -77,12 +78,15 @@ pub async fn shutdown_signal(
Ok(())
}
#[cfg(feature = "prometheus")]
use tokio::task::JoinHandle;
#[cfg(feature = "prometheus")]
pub async fn serve_metrics(
addr: SocketAddr,
mut rx: tokio::sync::broadcast::Receiver<()>,
ready_worker_endpoint: bool,
) -> Result<(), hyper::Error> {
) -> JoinHandle<()> {
use std::sync::atomic::Ordering;
use axum::{routing::get, Router};
@@ -104,13 +108,18 @@ pub async fn serve_metrics(
router
};
axum::Server::bind(&addr)
.serve(router.into_make_service())
.with_graceful_shutdown(async {
rx.recv().await.ok();
println!("Graceful shutdown of metrics");
})
.await
tokio::spawn(async move {
if let Err(e) = axum::Server::bind(&addr)
.serve(router.into_make_service())
.with_graceful_shutdown(async {
rx.recv().await.ok();
println!("Graceful shutdown of metrics");
})
.await
{
tracing::error!("Error serving metrics: {}", e);
}
})
}
async fn metrics() -> Result<String, Error> {
backend/windmill-common/src/worker.rs
+200
View File
@@ -0,0 +1,200 @@
use std::{collections::HashMap, sync::Arc};
use itertools::Itertools;
use regex::Regex;
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use crate::{error, global_settings::CUSTOM_TAGS_SETTING, DB};
lazy_static::lazy_static! {
pub static ref WORKER_GROUP: String = std::env::var("WORKER_GROUP").unwrap_or_else(|_| "default".to_string());
pub static ref DEFAULT_TAGS : Vec<String> = vec![
"deno".to_string(),
"python3".to_string(),
"go".to_string(),
"bash".to_string(),
"powershell".to_string(),
"nativets".to_string(),
"mysql".to_string(),
"graphql".to_string(),
"bun".to_string(),
"postgresql".to_string(),
"bigquery".to_string(),
"snowflake".to_string(),
"graphql".to_string(),
"dependency".to_string(),
"flow".to_string(),
"hub".to_string(),
"other".to_string()];
pub static ref WORKER_CONFIG: Arc<RwLock<WorkerConfig>> = Arc::new(RwLock::new(WorkerConfig {
worker_tags: Default::default(),
dedicated_worker: Default::default(),
}));
pub static ref CLOUD_HOSTED: bool = std::env::var("CLOUD_HOSTED").is_ok();
pub static ref CUSTOM_TAGS: Vec<String> = std::env::var("CUSTOM_TAGS")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect::<Vec<_>>()).unwrap_or_default();
pub static ref CUSTOM_TAGS_PER_WORKSPACE: Arc<RwLock<(Vec<String>, HashMap<String, Vec<String>>)>> = Arc::new(RwLock::new((vec![], HashMap::new())));
pub static ref ALL_TAGS: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
static ref CUSTOM_TAG_REGEX: Regex = Regex::new(r"^(\w+)\(((?:\w+)\+?)+\)$").unwrap();
}
pub async fn reload_custom_tags_setting(db: &DB) -> error::Result<()> {
let q = sqlx::query!(
"SELECT value FROM global_settings WHERE name = $1",
CUSTOM_TAGS_SETTING
)
.fetch_optional(db)
.await?;
let tags = if let Some(q) = q {
if let Ok(v) = serde_json::from_value::<Vec<String>>(q.value.clone()) {
v
} else {
tracing::error!(
"Could not parse custom tags setting as vec of strings, found: {:#?}",
&q.value
);
vec![]
}
} else {
CUSTOM_TAGS.clone()
};
let custom_tags = process_custom_tags(tags);
{
let l = CUSTOM_TAGS_PER_WORKSPACE.read().await;
if l.clone() == custom_tags {
tracing::info!("Custom tags setting unchanged, skipping update");
return Ok(());
} else {
tracing::info!("Custom tags setting changed, updating");
}
}
{
let mut l = CUSTOM_TAGS_PER_WORKSPACE.write().await;
*l = custom_tags.clone()
}
{
let mut l = ALL_TAGS.write().await;
*l = [
custom_tags.0.clone(),
custom_tags.1.keys().map(|x| x.to_string()).collect_vec(),
]
.concat();
}
Ok(())
// pub static ref CUSTOM_TAGS_PER_WORKSPACE: (Vec<String>, HashMap<String, Vec<String>>) = process_custom_tags(std::env::var("CUSTOM_TAGS")
// .ok());
// pub static ref ALL_TAGS: Vec<String> = [CUSTOM_TAGS_PER_WORKSPACE.0.clone(), CUSTOM_TAGS_PER_WORKSPACE.1.keys().map(|x| x.to_string()).collect_vec()].concat();
}
fn process_custom_tags(tags: Vec<String>) -> (Vec<String>, HashMap<String, Vec<String>>) {
let mut global = vec![];
let mut specific: HashMap<String, Vec<String>> = HashMap::new();
for e in tags {
if let Some(cap) = CUSTOM_TAG_REGEX.captures(&e) {
let tag = cap.get(1).unwrap().as_str().to_string();
let workspaces = cap.get(2).unwrap().as_str().split("+");
specific.insert(tag, workspaces.map(|x| x.to_string()).collect_vec());
} else {
global.push(e.to_string());
}
}
(global, specific)
}
pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
let wc = WORKER_CONFIG.read().await;
let tags = wc.worker_tags.as_slice();
sqlx::query!(
"INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
worker_instance,
worker_name,
ip,
tags,
*WORKER_GROUP,
wc.dedicated_worker.as_ref().map(|x| format!("{}:{}", x.workspace_id, x.path))
)
.execute(db)
.await
.expect("insert worker_ping initial value");
}
pub async fn load_worker_config(db: &DB) -> error::Result<WorkerConfig> {
let config: WorkerConfigOpt = sqlx::query_scalar!(
"SELECT config FROM worker_group_config WHERE name = $1",
*WORKER_GROUP
)
.fetch_optional(db)
.await?
.flatten()
.map(|x| serde_json::from_value(x).ok())
.flatten()
.unwrap_or_default();
let dedicated_worker = config.dedicated_worker.map(|x| {
let splitted = x.split(':').to_owned().collect_vec();
if splitted.len() != 2 {
panic!("DEDICATED_WORKER setting should be in the form of <workspace>:<script_path>")
} else {
let workspace = splitted[0];
let script_path = splitted[1];
WorkspacedPath { workspace_id: workspace.to_string(), path: script_path.to_string() }
}
});
Ok(WorkerConfig {
worker_tags: config
.worker_tags
.or_else(|| {
if let Some(ref dedicated_worker) = dedicated_worker.as_ref() {
Some(vec![format!(
"{}:{}",
dedicated_worker.workspace_id, dedicated_worker.path
)])
} else {
std::env::var("WORKER_TAGS")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect())
}
})
.unwrap_or_else(|| DEFAULT_TAGS.clone()),
dedicated_worker,
})
}
#[derive(Clone, PartialEq)]
pub struct WorkspacedPath {
pub workspace_id: String,
pub path: String,
}
#[derive(Serialize, Deserialize)]
pub struct WorkerConfigOpt {
pub worker_tags: Option<Vec<String>>,
pub dedicated_worker: Option<String>,
}
impl Default for WorkerConfigOpt {
fn default() -> Self {
Self { worker_tags: Default::default(), dedicated_worker: Default::default() }
}
}
#[derive(PartialEq)]
pub struct WorkerConfig {
pub worker_tags: Vec<String>,
pub dedicated_worker: Option<WorkspacedPath>,
}
backend/windmill-queue/src/jobs.rs
+14 -57
View File
@@ -14,7 +14,6 @@ use std::time::Instant;
use anyhow::Context;
use async_recursion::async_recursion;
use chrono::{DateTime, Duration, Utc};
use itertools::Itertools;
use reqwest::Client;
use rsmq_async::RsmqConnection;
use serde_json::json;
@@ -37,9 +36,13 @@ use windmill_common::{
schedule::{schedule_to_user, Schedule},
scripts::{ScriptHash, ScriptLang},
users::{username_to_permissioned_as, SUPERADMIN_SECRET_EMAIL},
worker::WORKER_CONFIG,
DB, METRICS_ENABLED,
};
#[cfg(feature = "enterprise")]
use windmill_common::worker::CLOUD_HOSTED;
use crate::{
schedule::{get_schedule_opt, push_scheduled_job},
QueueTransaction,
@@ -66,57 +69,6 @@ lazy_static::lazy_static! {
"Total number of jobs pulled from the queue."
)
.unwrap();
pub static ref CLOUD_HOSTED: bool = std::env::var("CLOUD_HOSTED").is_ok();
pub static ref DEFAULT_TAGS : Vec<String> = vec![
"deno".to_string(),
"python3".to_string(),
"go".to_string(),
"bash".to_string(),
"powershell".to_string(),
"nativets".to_string(),
"mysql".to_string(),
"graphql".to_string(),
"bun".to_string(),
"postgresql".to_string(),
"bigquery".to_string(),
"snowflake".to_string(),
"graphql".to_string(),
"dependency".to_string(),
"flow".to_string(),
"hub".to_string(),
"other".to_string()];
pub static ref DEDICATED_WORKER: Option<(String, String)> = std::env::var("DEDICATED_WORKER")
.ok()
.map(|x| {
let splitted = x.split(':').to_owned().collect_vec();
if splitted.len() != 2 {
panic!("DEDICATED_WORKER should be in the form of <workspace>:<script_path>")
} else {
let workspace = splitted[0];
let script_path = splitted[1];
(workspace.to_string(), script_path.to_string())
}
});
pub static ref ACCEPTED_TAGS: Vec<String> = {
let worker_tags = std::env::var("WORKER_TAGS")
.ok()
.map(|x| x.split(',').map(|x| x.to_string()).collect())
.unwrap_or_else(|| DEFAULT_TAGS.clone());
if let Some(ref dedicated_worker) = DEDICATED_WORKER.as_ref() {
vec![format!("{}:{}", dedicated_worker.0, dedicated_worker.1)]
} else {
worker_tags
}
};
pub static ref IS_WORKER_TAGS_DEFINED: bool = std::env::var("WORKER_TAGS").ok().is_some();
// When compiled in 'benchmark' mode, this flags is exposed via the /workers/toggle endpoint
// and make it possible to disable to current active workers (such that they don't pull any)
@@ -425,7 +377,7 @@ pub async fn add_completed_job<R: rsmq_async::RsmqConnection + Clone + Send>(
if !is_flow && _duration > 1000 {
let additional_usage = _duration / 1000;
let w_id = &queued_job.workspace_id;
let premium_workspace = *CLOUD_HOSTED
let premium_workspace = *windmill_common::worker::CLOUD_HOSTED
&& sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", w_id)
.fetch_one(db)
.await
@@ -1153,7 +1105,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
// TODO: REDIS: Race conditions / replace last_ping
// TODO: shuffle this list to have fairness
let mut all_tags = ACCEPTED_TAGS.clone();
let mut all_tags = WORKER_CONFIG.read().await.worker_tags.clone();
let mut msg: Option<_> = None;
let mut tag = None;
@@ -1210,6 +1162,8 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
* suspend_until is non-null
* and suspend = 0 when the resume messages are received
* or suspend_until <= now() if it has timed out */
let config = WORKER_CONFIG.read().await;
let tags = config.worker_tags.as_slice();
let r = if suspend_first {
sqlx::query_as::<_, QueuedJob>("UPDATE queue
@@ -1226,17 +1180,21 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
LIMIT 1
)
RETURNING *")
.bind(ACCEPTED_TAGS.as_slice())
.bind(tags)
.fetch_optional(db)
.await?
} else {
None
};
drop(config);
if r.is_none() {
// #[cfg(feature = "benchmark")]
// let instant = Instant::now();
let config = WORKER_CONFIG.read().await;
let tags = config.worker_tags.as_slice();
let r = sqlx::query_as::<_, QueuedJob>(
"UPDATE queue
SET running = true
@@ -1253,10 +1211,9 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
)
RETURNING *",
)
.bind(ACCEPTED_TAGS.as_slice())
.bind(tags)
.fetch_optional(db)
.await?;
// #[cfg(feature = "benchmark")]
// println!("pull query: {:?}", instant.elapsed());
backend/windmill-worker/src/common.rs
+1 -1
View File
@@ -5,12 +5,12 @@ use serde_json::{json, Value};
use sqlx::{Pool, Postgres};
use tokio::{fs::File, io::AsyncReadExt};
use windmill_api_client::{types::CreateResource, Client};
use windmill_common::worker::CLOUD_HOSTED;
use windmill_common::{
error::{self, Error},
jobs::QueuedJob,
variables::ContextualVariable,
};
use windmill_queue::CLOUD_HOSTED;
use anyhow::Result;
use std::{
backend/windmill-worker/src/config.rs
View File
backend/windmill-worker/src/lib.rs
+1 -1
View File
@@ -6,6 +6,7 @@ mod snowflake_executor;
mod bash_executor;
mod bun_executor;
mod common;
mod config;
mod dedicated_worker;
mod deno_executor;
mod global_cache;
@@ -17,5 +18,4 @@ mod pg_executor;
mod python_executor;
mod worker;
mod worker_flow;
pub use worker::*;
backend/windmill-worker/src/worker.rs
+58 -37
View File
@@ -28,12 +28,10 @@ use windmill_common::{
scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang},
users::SUPERADMIN_SECRET_EMAIL,
utils::{rd_string, StripPath},
worker::{update_ping, CLOUD_HOSTED, WORKER_CONFIG},
DB, IS_READY, METRICS_ENABLED,
};
use windmill_queue::{
canceled_job_to_result, get_queued_job, pull, ACCEPTED_TAGS, CLOUD_HOSTED, DEDICATED_WORKER,
HTTP_CLIENT, IS_WORKER_TAGS_DEFINED,
};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, HTTP_CLIENT};
use serde_json::{json, Value};
@@ -350,6 +348,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
_num_workers: u32,
ip: &str,
mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
killpill_tx: tokio::sync::broadcast::Sender<()>,
base_internal_url: &str,
rsmq: Option<R>,
_sync_barrier: Arc<RwLock<Option<Barrier>>>,
@@ -388,7 +387,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let mut last_ping = Instant::now() - Duration::from_secs(NUM_SECS_PING + 1);
insert_initial_ping(worker_instance, &worker_name, ip, db).await;
update_ping(worker_instance, &worker_name, ip, db).await;
let uptime_metric =
prometheus::register_counter!(WORKER_UPTIME_OPTS.clone().const_label("name", &worker_name))
@@ -611,17 +610,21 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
IS_READY.store(true, Ordering::Relaxed);
tracing::info!(worker = %worker_name, "listening for jobs");
let (dedicated_worker_tx, dedicated_worker_handle) = if let Some((_workspace, _script_path)) =
DEDICATED_WORKER.clone()
let (dedicated_worker_tx, dedicated_worker_handle) = if let Some(_wp) =
WORKER_CONFIG.read().await.dedicated_worker.clone()
{
#[cfg(not(feature = "enterprise"))]
panic!("Dedicated worker is an enterprise feature");
{
tracing::error!("Dedicated worker is an enterprise feature");
killpill_tx.send(()).expect("send");
return;
}
#[cfg(feature = "enterprise")]
{
let (dedicated_worker_tx, dedicated_worker_rx) =
mpsc::channel::<QueuedJob>(MAX_BUFFERED_DEDICATED_JOBS);
let killpill_rx = killpill_rx.resubscribe();
let mut killpill_rx = killpill_rx.resubscribe();
let db = db.clone();
let worker_dir = worker_dir.clone();
let base_internal_url = base_internal_url.to_string();
@@ -645,19 +648,52 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.execute(&db)
.await
{
panic!("failed to create token for dedicated worker: {:?}", e)
tracing::error!("failed to create token for dedicated worker: {:?}", e);
killpill_tx.send(()).expect("send");
};
let (content, lock, _language, envs) = sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>, Option<Vec<String>>)>(
let (content, lock, _language, envs) = {
let r;
loop {
let q = sqlx::query_as::<_, (String, Option<String>, Option<ScriptLang>, Option<Vec<String>>)>(
"SELECT content, lock, language, envs FROM script WHERE path = $1 AND workspace_id = $2 AND
created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2 AND
deleted = false AND lock IS not NULL AND lock_error_logs IS NULL)",
)
.bind(&_script_path)
.bind(&_workspace)
.bind(&_wp.path)
.bind(&_wp.workspace_id)
.fetch_optional(&db)
.await.expect("Failed to fetch script for dedicated worker")
.expect(&format!("Failed to fetch script `{_script_path}` in workspace {_workspace} for dedicated worker"));
.await;
if let Ok(q) = q {
if let Some(wp) = q {
r = wp;
break;
} else {
tracing::error!(
"Failed to fetch script `{}` in workspace {} for dedicated worker. Retrying in 10s.",
_wp.path,
_wp.workspace_id
);
tokio::select! {
biased;
_ = killpill_rx.recv() => {
tracing::info!("Killing dedicated worker while it was attempting to fetch script");
return;
}
_ = tokio::time::sleep(Duration::from_secs(10)) => {
continue;
}
}
}
} else {
tracing::error!("Failed to fetch script for dedicated worker");
killpill_tx.send(()).expect("send");
return;
}
}
r
};
let worker_envs = build_envs(envs).expect("failed to build envs");
if let Err(e) = start_worker(
@@ -668,8 +704,8 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
&job_dir,
&worker_name,
worker_envs,
&_workspace,
&_script_path,
&_wp.workspace_id,
&_wp.path,
&token,
job_completed_tx,
dedicated_worker_rx,
@@ -706,9 +742,13 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let copy_tx = _copy_to_bucket_tx.clone();
if last_ping.elapsed().as_secs() > NUM_SECS_PING {
let wc = WORKER_CONFIG.read().await;
let tags = wc.worker_tags.as_slice();
sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1 WHERE worker = $2",
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2 WHERE worker = $3",
jobs_executed,
tags,
&worker_name
)
.execute(db)
@@ -1269,25 +1309,6 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
tracing::error!(job_id = %job.id, "error handling job: {err:?} {} {} {}", job.id, job.workspace_id, job.created_by);
}
async fn insert_initial_ping(
worker_instance: &str,
worker_name: &str,
ip: &str,
db: &Pool<Postgres>,
) {
let tags = ACCEPTED_TAGS.clone();
sqlx::query!(
"INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags) VALUES ($1, $2, $3, $4) ON CONFLICT (worker) DO NOTHING",
worker_instance,
worker_name,
ip,
if *IS_WORKER_TAGS_DEFINED { Some(tags.as_slice()) } else { None }
)
.execute(db)
.await
.expect("insert worker_ping initial value");
}
fn extract_error_value(log_lines: &str, i: i32) -> serde_json::Value {
return json!({"message": format!("ExitCode: {i}, last log lines:\n{}", ANSI_ESCAPE_RE.replace_all(log_lines.trim(), "").to_string()), "name": "ExecutionErr"});
}
frontend/src/lib/components/PageHeader.svelte
+2 -2
View File
@@ -12,7 +12,7 @@
<span class="flex items-center space-x-2">
<h1 class="!text-2xl font-semibold leading-6 tracking-tight">{title}</h1>
{#if tooltip != '' || documentationLink}
<Tooltip light {documentationLink} scale={0.9} wrapperClass="flex items-center">
<Tooltip {documentationLink}>
{tooltip}
</Tooltip>
{/if}
@@ -21,7 +21,7 @@
<span class="flex items-center space-x-2">
<h2 class="!text-sm font-semibold">{title}</h2>
{#if tooltip != '' || documentationLink}
<Tooltip light {documentationLink} scale={0.9} wrapperClass="flex items-center">
<Tooltip {documentationLink}>
{tooltip}
</Tooltip>
{/if}
frontend/src/lib/components/ScriptBuilder.svelte
+2 -2
View File
@@ -483,7 +483,7 @@
>
<Toggle
disabled={!$enterpriseLicense ||
!isCloudHosted() ||
isCloudHosted() ||
script.language != Script.language.BUN}
size="xs"
checked={Boolean(script.dedicated_worker)}
@@ -680,7 +680,7 @@
>
{title}
{#if desc}
<Tooltip {documentationLink} class="mb-0.5 ml-1">
<Tooltip {documentationLink}>
{desc}
</Tooltip>
{/if}
frontend/src/lib/components/Tooltip.svelte
+8 -11
View File
@@ -1,25 +1,22 @@
<script lang="ts">
import { faInfoCircle } from '@fortawesome/free-solid-svg-icons'
import Icon from 'svelte-awesome'
import type { PopoverPlacement } from './Popover.model'
import Popover from './Popover.svelte'
import { ExternalLink } from 'lucide-svelte'
import { ExternalLink, InfoIcon } from 'lucide-svelte'
export let light = false
export let scale = 0.8
export let wrapperClass = ''
export let placement: PopoverPlacement | undefined = undefined
export let documentationLink: string | undefined = undefined
</script>
<Popover notClickable {placement} class={wrapperClass}>
<Icon
class="{light
? 'text-gray-400 dark:text-gray-200 hover:text-tertiary dark:hover:text-gray-300'
: ' text-tertiary dark:text-gray-300 hover:text-tertiary dark:hover:text-gray-400'} cursor-pointer transition-all font-thin flex h-4 p-0.5 w-4 justify-center items-center {$$props.class}"
data={faInfoCircle}
{scale}
/>
<div
class="inline-flex w-3 mx-0.5 {light
? 'text-tertiary-inverse'
: 'text-tertiary'} {$$props.class} relative"
>
<InfoIcon class="-bottom-0.5 absolute" size={16} />
</div>
<svelte:fragment slot="text">
<slot />
{#if documentationLink}
frontend/src/lib/components/WorkspaceGroup.svelte
+246
View File
@@ -0,0 +1,246 @@
<script lang="ts">
import { X } from 'lucide-svelte'
import { Button, Popup } from './common'
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
import { WorkerService } from '$lib/gen'
import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte'
import { createEventDispatcher } from 'svelte'
import { sendUserToast } from '$lib/toast'
import { enterpriseLicense, superadmin } from '$lib/stores'
import Tooltip from './Tooltip.svelte'
export let name: string
export let config:
| undefined
| {
dedicated_worker?: string
worker_tags?: string[]
}
let nconfig: any = config
? config.worker_tags != undefined || config.dedicated_worker != undefined
? config
: {
worker_tags: []
}
: {
worker_tags: []
}
const defaultTags = [
'deno',
'python3',
'go',
'bash',
'powershell',
'dependency',
'flow',
'hub',
'other',
'bun'
]
const nativeTags = ['nativets', 'postgresql', 'mysql', 'graphql', 'snowflake']
let newTag: string = ''
$: selected = nconfig?.dedicated_worker != undefined ? 'dedicated' : 'normal'
const dispatch = createEventDispatcher()
async function deleteWorkerGroup() {
await WorkerService.deleteWorkerGroup({ name })
dispatch('reload')
}
let dirty = false
let open = false
</script>
<ConfirmationModal
{open}
title="Delete worker group"
confirmationText="Remove"
on:canceled={() => {
open = false
}}
on:confirmed={async () => {
deleteWorkerGroup()
open = false
}}
>
<div class="flex flex-col w-full space-y-4">
<span>Are you sure you want to remove this worker nconfig?</span>
</div>
</ConfirmationModal>
<div class="flex gap-2 items-center"
><h4 class="py-4 truncate w-40">{name}</h4>
{#if $superadmin}
<Popup
floatingConfig={{ strategy: 'absolute', placement: 'bottom-end' }}
containerClasses="border rounded-lg shadow-lg p-4 bg-surface"
>
<svelte:fragment slot="button">
<Button color="light" size="xs" nonCaptureEvent={true}>
<div class="flex flex-row gap-1 items-center"
>{config == undefined ? 'create' : 'edit'} config</div
>
</Button>
</svelte:fragment>
<ToggleButtonGroup
{selected}
on:selected={(e) => {
dirty = true
if (nconfig == undefined) {
nconfig = {}
}
console.log(e.detail)
if (e.detail == 'dedicated') {
nconfig.dedicated_worker = ''
nconfig.worker_tags = undefined
} else {
nconfig.dedicated_worker = undefined
nconfig.worker_tags = []
}
}}
class="mb-4"
>
<ToggleButton
position="left"
value="normal"
size="sm"
label="Any jobs within worker tags"
/>
<ToggleButton
position="dedicated"
value="dedicated"
size="sm"
label="Dedicated to a script"
/>
</ToggleButtonGroup>
{#if selected == 'normal'}
{#if nconfig?.worker_tags != undefined}
<div class="flex flex-col gap-1 pb-2">
{#each nconfig.worker_tags as tag}
<div class="flex gap-1 items-center"
><div class="text-sm">- {tag}</div>
<button
class="z-10 rounded-full p-1 duration-200 hover:bg-gray-200"
aria-label="Remove item"
on:click|preventDefault|stopPropagation={() => {
if (nconfig != undefined) {
nconfig.worker_tags = nconfig?.worker_tags?.filter((t) => t != tag) ?? []
}
}}
>
<X size={14} />
</button></div
>
{/each}
</div>
<input type="text" placeholder="new tag" bind:value={newTag} />
<div class="mt-1" />
<Button
variant="contained"
color="blue"
size="xs"
disabled={newTag == '' || nconfig.worker_tags?.includes(newTag)}
on:click={() => {
if (nconfig != undefined) {
nconfig.worker_tags = [...(nconfig?.worker_tags ?? []), newTag.replaceAll(' ', '_')]
newTag = ''
dirty = true
}
}}
>
Add tag
</Button>
<div class="flex flex-wrap mt-2 items-center gap-1">
<Button
variant="contained"
color="light"
size="xs"
on:click={() => {
if (nconfig != undefined) {
nconfig.worker_tags = defaultTags.concat(nativeTags)
dirty = true
}
}}
>
Reset to all tags <Tooltip>{defaultTags.concat(nativeTags).join(', ')}</Tooltip>
</Button>
<Button
variant="contained"
color="light"
size="xs"
on:click={() => {
if (nconfig != undefined) {
nconfig.worker_tags = nativeTags
dirty = true
}
}}
>
Reset to all tags minus native ones <Tooltip>{defaultTags.join(', ')}</Tooltip>
</Button>
<Button
variant="contained"
color="light"
size="xs"
on:click={() => {
if (nconfig != undefined) {
nconfig.worker_tags = nativeTags
dirty = true
}
}}
>
Reset to native tags <Tooltip>{nativeTags.join(', ')}</Tooltip>
</Button>
</div>
{/if}
{:else if selected == 'dedicated'}
{#if nconfig?.dedicated_worker != undefined}
<input
placeholder="<workspace>:<script path>"
type="text"
on:change={() => {
dirty = true
}}
bind:value={nconfig.dedicated_worker}
/>
<p class="text-2xs text-tertiary max-w-md mt-2"
>Workers will get killed upon detecting this setting change. It is assumed they are in
an environment where the supervisor will restart them. Upon restart, they will pick the
new dedicated worker config.</p
>
{/if}
{/if}
<div class="mt-4" />
<div class="flex gap-1 items-center">
<Button
variant="contained"
color="dark"
size="xs"
on:click={async () => {
await WorkerService.updateWorkerGroup({ name, requestBody: nconfig })
sendUserToast(
'Setting configuration, it can take up to 30s to get propagated to all workers'
)
dispatch('reload')
}}
disabled={!dirty || !$enterpriseLicense}
>
Apply changes {#if !$enterpriseLicense}(ee only){/if}
</Button>
{#if !$enterpriseLicense}<Tooltip
>{selected == 'dedicated'
? 'Dedicated workers are an enterprise only feature'
: 'The Worker Group Manager UI is an enterprise only feature. However, workers can still have their WORKER_TAGS passed as env'}</Tooltip
>{/if}
</div>
</Popup>
{#if config}
<Button color="light" size="xs" on:click={() => (open = true)} btnClasses="text-red-400">
delete config
</Button>
{/if}
{/if}
</div>
frontend/src/lib/components/common/button/Button.svelte
+3 -5
View File
@@ -89,9 +89,9 @@
},
light: {
border:
'border bg-surface hover:bg-surface-hover focus:bg-surface-hover text-primary hover:text-secondary focus:text-secondary focus:ring-surface-selected',
'border bg-surface hover:bg-surface-hover focus:bg-surface-hover text-primary hover:text-secondary focus:text-secondary focus:ring-surface-selected',
contained:
'bg-surface hover:bg-surface-hover focus:bg-surface-hover text-primary focus:ring-surface-selected',
'bg-surface border-transparent hover:bg-surface-hover focus:bg-surface-hover text-primary focus:ring-surface-selected',
divider: 'divide-x divide-gray-200 dark:divide-gray-700'
}
}
@@ -174,9 +174,7 @@
on:blur
class={twMerge(
buttonClass,
disabled
? '!bg-surface-disabled !text-tertiary border border-disabled !cursor-not-allowed'
: ''
disabled ? '!bg-surface-disabled !text-tertiary border !cursor-not-allowed' : 'border'
)}
{id}
tabindex={disabled ? -1 : 0}
frontend/src/lib/components/flows/header/FlowPreviewButtons.svelte
+2
View File
@@ -26,6 +26,8 @@
'settings-same-worker',
'settings-graph',
'settings-worker-group',
'settings-cache',
'settings-concurrency',
'inputs',
'schedules',
'failure',
frontend/src/lib/utils.ts
+14 -13
View File
@@ -228,28 +228,29 @@ export function setQueryWithoutLoad(
}, bounceTime ?? 200)
}
export function groupBy<T>(
items: T[],
toGroup: (t: T) => string,
toSort: (t: T) => string,
dflts: string[] = []
): [string, T[]][] {
let r: Record<string, T[]> = {}
export function groupBy<K, V>(
items: V[],
toGroup: (t: V) => K,
toSort: (t: V) => string,
dflts: K[] = []
): [K, V[]][] {
let r: Map<K, V[]> = new Map()
for (const dflt of dflts) {
r[dflt] = []
r.set(dflt as K, [])
}
items.forEach((sc) => {
let section = toGroup(sc)
if (section in r) {
r[section].push(sc)
r[section].sort((a, b) => toSort(a).localeCompare(toSort(b)))
if (r.has(section)) {
let arr = r.get(section)!
arr.push(sc)
arr.sort((a, b) => toSort(a).localeCompare(toSort(b)))
} else {
r[section] = [sc]
r.set(section, [sc])
}
})
return Object.entries(r).sort((s1, s2) => {
return [...r.entries()].sort((s1, s2) => {
let n1 = s1[0]
let n2 = s2[0]
frontend/src/routes/(root)/(logged)/workers/+page.svelte
+273 -92
View File
@@ -1,6 +1,6 @@
<script lang="ts">
import CenteredPage from '$lib/components/CenteredPage.svelte'
import { Skeleton } from '$lib/components/common'
import { Button, Popup, Skeleton } from '$lib/components/common'
import Badge from '$lib/components/common/badge/Badge.svelte'
import PageHeader from '$lib/components/PageHeader.svelte'
import Cell from '$lib/components/table/Cell.svelte'
@@ -8,50 +8,84 @@
import Head from '$lib/components/table/Head.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import WorkspaceGroup from '$lib/components/WorkspaceGroup.svelte'
import { WorkerService, type WorkerPing, SettingService } from '$lib/gen'
import { enterpriseLicense, superadmin } from '$lib/stores'
import { sendUserToast } from '$lib/toast'
import { displayDate, groupBy } from '$lib/utils'
import { displayDate, groupBy, truncate } from '$lib/utils'
import { faPlus } from '@fortawesome/free-solid-svg-icons'
import { Loader2, Pen, X } from 'lucide-svelte'
import { onDestroy, onMount } from 'svelte'
let workers: WorkerPing[] | undefined = undefined
let filteredWorkers: WorkerPing[] = []
let groupedWorkers: [string, WorkerPing[]][] = []
let workerGroups: Record<string, any> | undefined = undefined
let groupedWorkers: [string, [[string, string], WorkerPing[]][]][] = []
let intervalId: NodeJS.Timer | undefined
let globalCache = false
let customTags: string[] | undefined = []
$: filteredWorkers = (workers ?? []).filter((x) => (x.last_ping ?? 0) < 300)
$: groupedWorkers = groupBy(
filteredWorkers,
(wp: WorkerPing) => wp.worker_instance,
(wp: WorkerPing) => wp.worker
groupBy(
filteredWorkers,
(wp: WorkerPing) => [wp.worker_instance, wp.worker_group],
(wp: WorkerPing) => wp.worker
),
(x) => x[0][1],
(x) => x[0][0]
)
const worker_s3_bucket_sync = 'worker_s3_bucket_sync'
const WORKER_S3_BUCKET_SYNC_SETTING = 'worker_s3_bucket_sync'
const CUSTOM_TAGS_SETTING = 'custom_tags'
let timeSinceLastPing = 0
async function loadWorkers(): Promise<void> {
try {
workers = await WorkerService.listWorkers({ perPage: 100 })
workers = await WorkerService.listWorkers({ perPage: 1000 })
timeSinceLastPing = 0
} catch (err) {
sendUserToast(`Could not load workers: ${err}`, true)
}
}
async function loadWorkerGroups(): Promise<void> {
try {
workerGroups = Object.fromEntries(
(await WorkerService.listWorkerGroups()).map((x) => [x.name, x.config])
)
} catch (err) {
sendUserToast(`Could not load workers: ${err}`, true)
}
}
let secondInterval: NodeJS.Timer | undefined = undefined
onMount(() => {
loadWorkers()
intervalId = setInterval(loadWorkers, 5000)
loadWorkerGroups()
intervalId = setInterval(() => {
loadWorkers()
loadWorkerGroups()
}, 5000)
secondInterval = setInterval(() => {
timeSinceLastPing += 1
}, 1000)
loadGlobalCache()
loadCustomTags()
})
async function loadGlobalCache() {
try {
globalCache = (await SettingService.getGlobal({ key: worker_s3_bucket_sync })) ?? true
globalCache = (await SettingService.getGlobal({ key: WORKER_S3_BUCKET_SYNC_SETTING })) ?? true
} catch (err) {
sendUserToast(`Could not load global cache: ${err}`, true)
}
}
async function loadCustomTags() {
try {
customTags = (await SettingService.getGlobal({ key: CUSTOM_TAGS_SETTING })) ?? []
} catch (err) {
sendUserToast(`Could not load global cache: ${err}`, true)
}
@@ -65,102 +99,249 @@
clearInterval(secondInterval)
}
})
let newGroupName = ''
async function addGroup() {
await WorkerService.updateWorkerGroup({ name: newGroupName, requestBody: {} })
loadWorkerGroups()
}
let newTag: string = ''
</script>
<CenteredPage>
<PageHeader
title="Workers"
tooltip="The workers are the dutiful servants that execute your scripts.
This page enables you to know their IP in case you need whitelisting and also display liveness information"
tooltip="The workers are the dutiful servants that execute the jobs."
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups"
/>
>
{#if $superadmin}
<div class="flex flex-row-reverse w-full pb-2 items-center gap-4">
<div class="flex gap-2 items-center">
<Toggle
checked={globalCache}
on:change={async (e) => {
try {
console.log('Setting global cache to', e.detail)
await SettingService.setGlobal({
key: WORKER_S3_BUCKET_SYNC_SETTING,
requestBody: { value: e.detail }
})
globalCache = e.detail
} catch (err) {
sendUserToast(`Could not set global cache: ${err}`, true)
}
}}
options={{ right: 'global cache to s3' }}
disabled={!$enterpriseLicense}
/>
<Tooltip
><p
>global cache to s3 is an enterprise feature that enable workers to do fast cold start
and share a single cache backed by s3 to ensure that even with a high number of
workers, dependencies for python/deno/bun/go are only downloaded for the first time
only once by the whole fleet.
</p>require S3_CACHE_BUCKET to be set and has NO effect otherwise (even if this setting
is on)</Tooltip
>
</div>
<div
><Popup
floatingConfig={{ strategy: 'absolute', placement: 'bottom-end' }}
containerClasses="border rounded-lg shadow-lg p-4 bg-surface"
>
<svelte:fragment slot="button">
<Button color="dark" size="xs" nonCaptureEvent={true}>
<div class="flex flex-row gap-1 items-center"
><Pen size={14} /> Assignable tags&nbsp;<Tooltip light
>Tags are assigned to scripts and flows. Workers only accept jobs that
correspond to their worker tags. Scripts have a default tag based on the
language they are in but users can choose to override their tags with custom
ones. This editor allow you to set the custom tags one can override the scripts
and flows with.</Tooltip
></div
>
</Button>
</svelte:fragment>
<div class="flex flex-col w-72 p-2 gap-2">
{#if customTags == undefined}
<Loader2 class="animate-spin" />
{:else}
<div class="flex flex-col">
{#each customTags as customTag}
<div class="font-mono flex items-center gap-2 w-full">
<div class="w-full">- {customTag}</div>
<button
class="z-10 rounded-full p-1 duration-200 hover:bg-gray-200"
aria-label="Remove item"
on:click|preventDefault|stopPropagation={async () => {
await SettingService.setGlobal({
key: CUSTOM_TAGS_SETTING,
requestBody: { value: customTags?.filter((x) => x != customTag) }
})
loadCustomTags()
sendUserToast('Tag removed')
}}
>
<X size={14} />
</button>
</div>
{/each}
</div>
<input type="text" bind:value={newTag} />
<Button
variant="contained"
color="blue"
size="sm"
on:click={async () => {
await SettingService.setGlobal({
key: CUSTOM_TAGS_SETTING,
requestBody: {
value: [...(customTags ?? []), newTag.trim().replaceAll(' ', '_')]
}
})
loadCustomTags()
sendUserToast('Tag added')
}}
disabled={newTag.trim() == ''}
>
Add
</Button>
<span class="text-2xs text-tertiary"
>For tags specific to some workspaces, use <pre class="inline"
>tag(workspace1+workspace2)</pre
></span
>
<span class="text-2xs text-tertiary"
>For dynamic tags based on the workspace, use <pre class="inline">$workspace</pre
>, e.g:
<pre class="inline">tag-$workspace</pre></span
>
{/if}
</div>
</Popup>
</div>
</div>
{/if}
</PageHeader>
{#if $superadmin}
<div class="flex flex-row-reverse w-full pb-2 items-center gap-2">
<Tooltip
><p
>global cache to s3 is an enterprise feature that enable workers to do fast cold start and
share a single cache backed by s3 to ensure that even with a high number of workers,
dependencies for python/deno/bun/go are only downloaded for the first time only once by
the whole fleet.
</p>require S3_CACHE_BUCKET to be set and has NO effect otherwise (even if this setting is
on)</Tooltip
>
<Toggle
checked={globalCache}
on:change={async (e) => {
try {
console.log('Setting global cache to', e.detail)
await SettingService.setGlobal({
key: worker_s3_bucket_sync,
requestBody: { value: e.detail }
})
globalCache = e.detail
} catch (err) {
sendUserToast(`Could not set global cache: ${err}`, true)
}
}}
options={{ right: 'global cache to s3' }}
disabled={!$enterpriseLicense}
/>
</div>
{/if}
{#if workers != undefined}
{#if groupedWorkers.length == 0}
<p>No workers seems to be available</p>
{/if}
<DataTable>
<Head>
<tr>
<Cell head first>Worker</Cell>
<Cell head>
<div class="flex flex-row items-center gap-1">
Custom Tags
<Tooltip
light
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups#assign-custom-worker-groups"
>
If defined, the workers only pull jobs with the same corresponding tag
</Tooltip>
</div>
</Cell>
<Cell head>Last ping</Cell>
<Cell head>Worker start</Cell>
<Cell head>Nb of jobs executed</Cell>
<Cell head last>Liveness</Cell>
</tr>
</Head>
<tbody>
{#each groupedWorkers as [section, workers]}
<tr class="border-t">
<Cell first colspan="6" scope="colgroup" class="bg-surface-secondary/60 py-2 border-b">
Instance: <Badge color="gray">{section}</Badge>
IP: <Badge color="gray">{workers[0].ip}</Badge>
</Cell>
</tr>
<div class="py-4 w-full flex justify-between"
><h2
>Worker Groups <Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups"
>Worker groups are groups of workers that share a config and are meant to be identical.
Worker groups are meant to be used with tags. Tags can be assigned to scripts and flows
and can be seen as dedicated queues. Only the corresponding
</Tooltip></h2
>
<div />
{#if $superadmin}
<div class="flex flex-row items-center">
<input class="mr-2 h-full" placeholder="New group name" bind:value={newGroupName} />
<Button
size="sm"
startIcon={{ icon: faPlus }}
disabled={!newGroupName}
on:click={addGroup}
>
New worker group config
</Button>
<Tooltip>Worker Group configs are propagated to every workers in the worker group</Tooltip
>
</div>
{/if}</div
>
{#each groupedWorkers as worker_group}
<WorkspaceGroup
name={worker_group[0]}
config={(workerGroups ?? {})[worker_group[0]]}
on:reload={() => {
loadWorkerGroups()
}}
/>
{#if workers}
{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed }}
<tr>
<Cell first>{worker}</Cell>
<Cell>{custom_tags?.join(', ') ?? ''}</Cell>
<Cell>{last_ping != undefined ? last_ping + timeSinceLastPing : -1}s ago</Cell>
<Cell>{displayDate(started_at)}</Cell>
<Cell>{jobs_executed}</Cell>
<Cell last>
<Badge
color={last_ping != undefined ? (last_ping < 60 ? 'green' : 'red') : 'gray'}
<DataTable>
<Head>
<tr>
<Cell head first>Worker</Cell>
<Cell head>
<div class="flex flex-row items-center gap-1">
Worker Tags
<Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups#assign-custom-worker-groups"
>
If defined, the workers only pull jobs with the same corresponding tag
</Tooltip>
</div>
</Cell>
<Cell head>Last ping</Cell>
<Cell head>Worker start</Cell>
<Cell head>Nb of jobs executed</Cell>
<Cell head last>Liveness</Cell>
</tr>
</Head>
<tbody>
{#each worker_group[1] as [section, workers]}
<tr class="border-t">
<Cell
first
colspan="6"
scope="colgroup"
class="bg-surface-secondary/60 py-2 border-b"
>
Instance: <Badge color="gray">{section[0]}</Badge>
IP: <Badge color="gray">{workers[0].ip}</Badge>
</Cell>
</tr>
{#if workers}
{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed }}
<tr>
<Cell first>{worker}</Cell>
<Cell
>{#if custom_tags && custom_tags?.length > 2}{truncate(
custom_tags?.join(', ') ?? '',
10
)}
<Tooltip>{custom_tags?.join(', ')}</Tooltip>{:else}{custom_tags?.join(', ') ??
''}{/if}</Cell
>
{last_ping != undefined ? (last_ping < 60 ? 'Alive' : 'Dead') : 'Unknown'}
</Badge>
</Cell>
</tr>
{/each}
{/if}
{/each}
</tbody>
</DataTable>
<Cell>{last_ping != undefined ? last_ping + timeSinceLastPing : -1}s ago</Cell>
<Cell>{displayDate(started_at)}</Cell>
<Cell>{jobs_executed}</Cell>
<Cell last>
<Badge
color={last_ping != undefined ? (last_ping < 60 ? 'green' : 'red') : 'gray'}
>
{last_ping != undefined ? (last_ping < 60 ? 'Alive' : 'Dead') : 'Unknown'}
</Badge>
</Cell>
</tr>
{/each}
{/if}
{/each}
</tbody>
</DataTable>
<div class="pb-4" />
{/each}
{#each Object.entries(workerGroups ?? {}).filter((x) => !groupedWorkers.some((y) => y[0] == x[0])) as worker_group}
<WorkspaceGroup
on:reload={() => {
loadWorkerGroups()
}}
name={worker_group[0]}
config={worker_group[1]}
/>
<div class="text-xs text-tertiary"> No workers currently in this worker group </div>
{/each}
{:else}
<div class="flex flex-col">
{#each new Array(4) as _}