mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-20 00:03:21 +00:00
feat: route dev workspace and fork jobs to their own worker groups (#11811)
* feat: route dev workspace and fork jobs to their own worker groups * fix: harden dev workspace isolation against review findings * fix: verify hashed dedicated tags and keep dev lineage across renames * fix: let tag-scoped tokens reach jobs queued on dev workspace tags * fix: expand tag scopes only in dev workspaces and forks * fix: expand tag scopes only while dev workspace isolation is on Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat: collapse the dev workspace isolation section by default Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * feat: count dev workspace isolation toggles and dev-serving group flags Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: validate the dev workspace tag suffix on every instance config writer Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * chore: update ee-repo-ref to 71e9c0de8317e839f955ebef6a819062f20ac418 This commit updates the EE repository reference after PR #892 was merged in windmill-ee-private. Previous ee-repo-ref: b5ab84e1705f939c399f6315a03f88b55bce1804 New ee-repo-ref: 71e9c0de8317e839f955ebef6a819062f20ac418 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5.5
windmill-internal-app[bot]
Claude Opus 5.5
windmill-internal-app[bot]
parent
a52154877c
commit
6e565b0dcb
+28
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\n WITH RECURSIVE chain AS (\n SELECT id, parent_workspace_id, is_dev_workspace, 0 AS depth\n FROM workspace WHERE id = $1\n UNION ALL\n SELECT w.id, w.parent_workspace_id, w.is_dev_workspace, chain.depth + 1\n FROM workspace w\n JOIN chain ON w.id = chain.parent_workspace_id\n WHERE chain.depth < 20\n )\n SELECT\n (SELECT id FROM chain\n WHERE parent_workspace_id IS NULL\n OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%')\n ORDER BY depth LIMIT 1) AS id,\n (SELECT parent_workspace_id IS NOT NULL FROM chain WHERE depth = 0) AS has_parent\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "id",
|
||||
"type_info": "Varchar"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "has_parent",
|
||||
"type_info": "Bool"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null,
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "043cd99b3da09bca6cd2cf1e0adcd674de140b7f93897286447279e74777e99e"
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT config FROM config WHERE name LIKE 'worker\\_\\_%'",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "config",
|
||||
"type_info": "Jsonb"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": []
|
||||
},
|
||||
"nullable": [
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "0fb0fea54a1580d4b7ffd37ca43e169f51dbbdc3584d3522f25e97e180e307fd"
|
||||
}
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT config FROM config WHERE name = $1 FOR UPDATE",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "config",
|
||||
"type_info": "Jsonb"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "2d22fe27288580cb727e43b2abe3be2e6e6e3b03e61ce8f2879627115224c8cf"
|
||||
}
|
||||
Generated
+1
@@ -16229,6 +16229,7 @@ dependencies = [
|
||||
"uuid",
|
||||
"windmill-api-auth",
|
||||
"windmill-common",
|
||||
"windmill-queue",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -1 +1 @@
|
||||
664708fff04d4f9e0660bcae3edb307fb2c9481b
|
||||
71e9c0de8317e839f955ebef6a819062f20ac418
|
||||
|
||||
+16
-8
@@ -44,7 +44,8 @@ use windmill_common::{
|
||||
CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING,
|
||||
CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
|
||||
CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, DEFAULT_TAGS_WORKSPACES_SETTING,
|
||||
DEPENDENCY_JOB_TAG_SETTING, DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING,
|
||||
DEPENDENCY_JOB_TAG_SETTING, DEV_WORKSPACE_ISOLATION_SETTING,
|
||||
DEV_WORKSPACE_TAG_SUFFIX_SETTING, DISABLE_PASSWORD_LOGIN_SETTING, EMAIL_DOMAIN_SETTING,
|
||||
ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
|
||||
EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING,
|
||||
HTTP_ROUTE_DEFAULT_ALLOWED_ORIGINS_SETTING, HTTP_ROUTE_WORKSPACED_ROUTE_SETTING,
|
||||
@@ -129,7 +130,8 @@ use windmill_worker::{
|
||||
|
||||
use crate::monitor::{
|
||||
initial_load, load_concurrency_key_max_queued, load_dependency_job_tag,
|
||||
load_disable_password_login, load_fork_workspace_tag_append_fork_suffix, load_keep_job_dir,
|
||||
load_dev_workspace_isolation, load_dev_workspace_tag_suffix, load_disable_password_login,
|
||||
load_fork_workspace_tag_append_fork_suffix, load_keep_job_dir,
|
||||
load_mcp_disable_token_query_param, load_metrics_debug_enabled, load_preview_tags_override,
|
||||
load_require_preexisting_user, load_retention_period_overrides, load_tag_per_workspace_enabled,
|
||||
load_tag_per_workspace_workspaces, load_workspace_fairness_duration_secs,
|
||||
@@ -1052,12 +1054,8 @@ Windmill Community Edition {GIT_VERSION}
|
||||
|
||||
// Before the first load: it can sign a token for instance object storage.
|
||||
#[cfg(all(feature = "enterprise", feature = "openidconnect", feature = "private"))]
|
||||
windmill_common::oidc_ee::init_oidc_signing(
|
||||
&conn,
|
||||
server_mode,
|
||||
killpill_rx.resubscribe(),
|
||||
)
|
||||
.await?;
|
||||
windmill_common::oidc_ee::init_oidc_signing(&conn, server_mode, killpill_rx.resubscribe())
|
||||
.await?;
|
||||
|
||||
initial_load(
|
||||
&conn,
|
||||
@@ -1881,6 +1879,16 @@ async fn process_notify_event(
|
||||
tracing::error!("Error loading dependency job tag: {e:#}");
|
||||
}
|
||||
}
|
||||
DEV_WORKSPACE_ISOLATION_SETTING => {
|
||||
if let Err(e) = load_dev_workspace_isolation(db).await {
|
||||
tracing::error!("Error loading dev workspace isolation: {e:#}");
|
||||
}
|
||||
}
|
||||
DEV_WORKSPACE_TAG_SUFFIX_SETTING => match load_dev_workspace_tag_suffix(db).await {
|
||||
Ok(true) if worker_mode => reload_worker_config(db, tx.clone(), true).await,
|
||||
Ok(_) => (),
|
||||
Err(e) => tracing::error!("Error loading dev workspace tag suffix: {e:#}"),
|
||||
},
|
||||
WORKSPACE_FAIRNESS_ENABLED_SETTING => {
|
||||
if let Err(e) = load_workspace_fairness_enabled(db).await {
|
||||
tracing::error!("Error loading workspace fairness enabled: {e:#}");
|
||||
|
||||
+49
-6
@@ -60,7 +60,8 @@ use windmill_common::{
|
||||
CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERTS_ON_TOKEN_EXPIRY_SETTING,
|
||||
CRITICAL_ALERT_MUTE_UI_SETTING, CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART_SETTING,
|
||||
CRITICAL_ERROR_CHANNELS_SETTING, CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
|
||||
DEFAULT_TAGS_WORKSPACES_SETTING, DEPENDENCY_JOB_TAG_SETTING, DISABLE_PASSWORD_LOGIN,
|
||||
DEFAULT_TAGS_WORKSPACES_SETTING, DEPENDENCY_JOB_TAG_SETTING,
|
||||
DEV_WORKSPACE_ISOLATION_SETTING, DEV_WORKSPACE_TAG_SUFFIX_SETTING, DISABLE_PASSWORD_LOGIN,
|
||||
DISABLE_PASSWORD_LOGIN_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
|
||||
EXTRA_PIP_INDEX_URL_SETTING, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING,
|
||||
HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING,
|
||||
@@ -89,11 +90,13 @@ use windmill_common::{
|
||||
users::truncate_token,
|
||||
utils::{empty_as_none, now_from_db, report_critical_error, Mode, HUB_API_SECRET},
|
||||
worker::{
|
||||
load_env_vars, load_init_bash_from_env, load_periodic_bash_script_from_env,
|
||||
load_periodic_bash_script_interval_from_env, load_whitelist_env_vars_from_env,
|
||||
load_worker_config, store_pull_query, store_suspended_pull_query, Connection, WorkerConfig,
|
||||
CLOUD_HOSTED, CONCURRENCY_KEY_MAX_QUEUED, CONCURRENCY_KEY_MAX_QUEUED_DEFAULT,
|
||||
DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, DEPENDENCY_JOB_TAG,
|
||||
is_valid_dev_workspace_tag_suffix, load_env_vars, load_init_bash_from_env,
|
||||
load_periodic_bash_script_from_env, load_periodic_bash_script_interval_from_env,
|
||||
load_whitelist_env_vars_from_env, load_worker_config, store_pull_query,
|
||||
store_suspended_pull_query, Connection, WorkerConfig, CLOUD_HOSTED,
|
||||
CONCURRENCY_KEY_MAX_QUEUED, CONCURRENCY_KEY_MAX_QUEUED_DEFAULT,
|
||||
DEFAULT_DEV_WORKSPACE_TAG_SUFFIX, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES,
|
||||
DEPENDENCY_JOB_TAG, DEV_WORKSPACE_ISOLATION, DEV_WORKSPACE_TAG_SUFFIX,
|
||||
FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, INDEXER_CONFIG, PREVIEW_TAGS_OVERRIDE, SMTP_CONFIG,
|
||||
WINDMILL_DIR, WORKER_CONFIG, WORKER_GROUP, WORKSPACE_FAIRNESS_DURATION_SECS,
|
||||
WORKSPACE_FAIRNESS_ENABLED, WORKSPACE_FAIRNESS_MAX_PERCENT, WORKSPACE_FAIRNESS_MIN_TOTAL,
|
||||
@@ -321,6 +324,13 @@ pub async fn initial_load(
|
||||
pass.setting(DEPENDENCY_JOB_TAG_SETTING, false, |v| async move {
|
||||
apply_dependency_job_tag(v)
|
||||
});
|
||||
// Ahead of the worker config reload below, which expands the tags with it.
|
||||
pass.setting(DEV_WORKSPACE_TAG_SUFFIX_SETTING, false, |v| async move {
|
||||
apply_dev_workspace_tag_suffix(v);
|
||||
});
|
||||
pass.setting(DEV_WORKSPACE_ISOLATION_SETTING, false, |v| async move {
|
||||
apply_dev_workspace_isolation(v)
|
||||
});
|
||||
|
||||
// Load per-workspace retention overrides before the first cleanup tick so a fresh server
|
||||
// never sweeps globally without honoring configured longer-retention workspaces.
|
||||
@@ -1127,6 +1137,39 @@ pub fn apply_fork_workspace_tag_append_fork_suffix(value: Option<serde_json::Val
|
||||
};
|
||||
}
|
||||
|
||||
/// Returns whether the suffix changed, in which case a worker must reload its config: a group
|
||||
/// serving dev workspaces listens to its tags expanded with it.
|
||||
pub async fn load_dev_workspace_tag_suffix(db: &DB) -> error::Result<bool> {
|
||||
let v = load_value_from_global_settings(db, DEV_WORKSPACE_TAG_SUFFIX_SETTING).await?;
|
||||
Ok(apply_dev_workspace_tag_suffix(v))
|
||||
}
|
||||
|
||||
pub fn apply_dev_workspace_tag_suffix(value: Option<serde_json::Value>) -> bool {
|
||||
let suffix = match value {
|
||||
Some(serde_json::Value::String(t)) if !t.trim().is_empty() => t.trim().to_string(),
|
||||
_ => DEFAULT_DEV_WORKSPACE_TAG_SUFFIX.to_string(),
|
||||
};
|
||||
if !is_valid_dev_workspace_tag_suffix(&suffix) {
|
||||
tracing::error!("Ignoring invalid dev workspace tag suffix {suffix:?}");
|
||||
return false;
|
||||
}
|
||||
let prev = DEV_WORKSPACE_TAG_SUFFIX.swap(std::sync::Arc::new(suffix.clone()));
|
||||
*prev != suffix
|
||||
}
|
||||
|
||||
pub async fn load_dev_workspace_isolation(db: &DB) -> error::Result<()> {
|
||||
let v = load_value_from_global_settings(db, DEV_WORKSPACE_ISOLATION_SETTING).await?;
|
||||
apply_dev_workspace_isolation(v);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn apply_dev_workspace_isolation(value: Option<serde_json::Value>) {
|
||||
DEV_WORKSPACE_ISOLATION.store(
|
||||
matches!(value, Some(serde_json::Value::Bool(true))),
|
||||
Ordering::Relaxed,
|
||||
);
|
||||
}
|
||||
|
||||
pub async fn reload_critical_alert_mute_ui_setting(conn: &Connection) -> error::Result<()> {
|
||||
let v = load_value_from_global_settings_with_conn(conn, CRITICAL_ALERT_MUTE_UI_SETTING, true)
|
||||
.await?;
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
//! Sets the process-wide `DEV_WORKSPACE_ISOLATION`, so it lives in its own test binary: any other
|
||||
//! test pushing a job in the same process would land on a suffixed tag.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use sqlx::{Pool, Postgres};
|
||||
use windmill_common::{
|
||||
jobs::{JobPayload, RawCode},
|
||||
runnable_settings::{ConcurrencySettings, DebouncingSettings},
|
||||
scripts::{ScriptHash, ScriptLang},
|
||||
worker::DEV_WORKSPACE_ISOLATION,
|
||||
};
|
||||
use windmill_queue::{PushArgs, PushIsolationLevel};
|
||||
|
||||
async fn pushed_tag(
|
||||
db: &Pool<Postgres>,
|
||||
workspace_id: &str,
|
||||
payload: JobPayload,
|
||||
tag: Option<&str>,
|
||||
) -> String {
|
||||
try_pushed_tag(db, workspace_id, payload, tag)
|
||||
.await
|
||||
.expect("push must succeed")
|
||||
}
|
||||
|
||||
async fn try_pushed_tag(
|
||||
db: &Pool<Postgres>,
|
||||
workspace_id: &str,
|
||||
payload: JobPayload,
|
||||
tag: Option<&str>,
|
||||
) -> Result<String, windmill_common::error::Error> {
|
||||
let args = HashMap::new();
|
||||
let (id, tx) = windmill_queue::push(
|
||||
db,
|
||||
PushIsolationLevel::IsolatedRoot(db.clone()),
|
||||
workspace_id,
|
||||
payload,
|
||||
PushArgs::from(&args),
|
||||
"test-user",
|
||||
"test@windmill.dev",
|
||||
"u/test-user".to_string(),
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
false,
|
||||
false,
|
||||
None,
|
||||
true,
|
||||
tag.map(str::to_string),
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
false,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await.unwrap();
|
||||
Ok(sqlx::query_scalar("SELECT tag FROM v2_job WHERE id = $1")
|
||||
.bind(id)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap())
|
||||
}
|
||||
|
||||
fn preview() -> JobPayload {
|
||||
JobPayload::Code(RawCode {
|
||||
hash: None,
|
||||
content: "echo hi".to_string(),
|
||||
path: None,
|
||||
language: ScriptLang::Bash,
|
||||
lock: None,
|
||||
concurrency_settings: ConcurrencySettings::default().into(),
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
cache_ttl: None,
|
||||
cache_ignore_s3_path: None,
|
||||
dedicated_worker: None,
|
||||
modules: None,
|
||||
tag: None,
|
||||
})
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn dev_workspace_and_fork_jobs_run_on_suffixed_tags(db: Pool<Postgres>) {
|
||||
DEV_WORKSPACE_ISOLATION.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
// A dev workspace, a fork of it, and a fork whose parent was deleted (`ON DELETE SET NULL`).
|
||||
sqlx::query(
|
||||
"INSERT INTO workspace (id, name, owner, parent_workspace_id, is_dev_workspace) VALUES
|
||||
('staging', 'staging', 'test-user', 'test-workspace', true),
|
||||
('wm-fork-a', 'fork', 'test-user', 'staging', false),
|
||||
('wm-fork-orphan', 'orphan', 'test-user', NULL, false)",
|
||||
)
|
||||
.execute(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO workspace_settings (workspace_id)
|
||||
VALUES ('staging'), ('wm-fork-a'), ('wm-fork-orphan')",
|
||||
)
|
||||
.execute(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
for (workspace, tag, expected) in [
|
||||
("test-workspace", None, "bash"),
|
||||
("test-workspace", Some("gpu"), "gpu"),
|
||||
("staging", None, "bash-dev"),
|
||||
("staging", Some("gpu"), "gpu-dev"),
|
||||
// A tag inherited from a dev job (a flow step, an agent's tool) is not suffixed twice.
|
||||
("staging", Some("gpu-dev"), "gpu-dev"),
|
||||
("wm-fork-a", None, "bash-dev"),
|
||||
("wm-fork-orphan", None, "bash-dev"),
|
||||
] {
|
||||
assert_eq!(
|
||||
pushed_tag(&db, workspace, preview(), tag).await,
|
||||
expected,
|
||||
"{workspace} with tag {tag:?}"
|
||||
);
|
||||
}
|
||||
|
||||
// A dedicated worker tag already names the workspace, and its worker is placed by hand.
|
||||
let dedicated = JobPayload::ScriptHash {
|
||||
hash: ScriptHash(123412),
|
||||
path: "f/system/hello".to_string(),
|
||||
cache_ttl: None,
|
||||
cache_ignore_s3_path: None,
|
||||
dedicated_worker: Some(true),
|
||||
language: ScriptLang::Bun,
|
||||
priority: None,
|
||||
apply_preprocessor: false,
|
||||
concurrency_settings: ConcurrencySettings::default(),
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
labels: None,
|
||||
job_token_scopes: None,
|
||||
};
|
||||
assert_eq!(
|
||||
pushed_tag(&db, "staging", dedicated, None).await,
|
||||
"staging:f/system/hello"
|
||||
);
|
||||
|
||||
// A hashed dedicated tag keeps only a prefix of its workspace id: a dev workspace sharing that
|
||||
// prefix does not get to use another workspace's dedicated worker.
|
||||
let root = "a".repeat(40);
|
||||
let dev = format!("{root}-dev");
|
||||
sqlx::query(
|
||||
"INSERT INTO workspace (id, name, owner, parent_workspace_id, is_dev_workspace)
|
||||
VALUES ($1, 'root', 'test-user', NULL, false), ($2, 'dev', 'test-user', $1, true)",
|
||||
)
|
||||
.bind(&root)
|
||||
.bind(&dev)
|
||||
.execute(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query("INSERT INTO workspace_settings (workspace_id) VALUES ($1), ($2)")
|
||||
.bind(&root)
|
||||
.bind(&dev)
|
||||
.execute(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
let root_dedicated =
|
||||
windmill_common::worker::dedicated_worker_tag(&root, "f/some/long/script/path");
|
||||
assert!(
|
||||
try_pushed_tag(&db, &dev, preview(), Some(&root_dedicated))
|
||||
.await
|
||||
.is_err(),
|
||||
"{root_dedicated} must not stay unsuffixed, and suffixed it no longer fits a job tag"
|
||||
);
|
||||
}
|
||||
@@ -1533,6 +1533,32 @@ async fn declarative_sync_rejects_an_unusable_instance_banner(db: Pool<Postgres>
|
||||
);
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn declarative_sync_rejects_an_unusable_dev_workspace_tag_suffix(db: Pool<Postgres>) {
|
||||
clear_settings_and_configs(&db).await;
|
||||
let before = count_global_settings(&db).await;
|
||||
|
||||
let mut desired = BTreeMap::new();
|
||||
desired.insert(
|
||||
"dev_workspace_tag_suffix".to_string(),
|
||||
serde_json::json!("-dev' OR '1'='1"),
|
||||
);
|
||||
|
||||
let err = windmill_common::instance_config::sync_global_settings_declarative(
|
||||
&db,
|
||||
&BTreeMap::new(),
|
||||
&desired,
|
||||
&(),
|
||||
)
|
||||
.await
|
||||
.expect_err("a suffix the workers would ignore must fail the sync");
|
||||
assert!(
|
||||
err.to_string().contains("dev_workspace_tag_suffix"),
|
||||
"the error should name the offending setting, got: {err}"
|
||||
);
|
||||
assert_eq!(count_global_settings(&db).await, before);
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn declarative_sync_rejects_a_malformed_max_token_expiration(db: Pool<Postgres>) {
|
||||
clear_settings_and_configs(&db).await;
|
||||
|
||||
@@ -6,6 +6,7 @@ use axum::{
|
||||
};
|
||||
use chrono::TimeZone;
|
||||
use http::{request::Parts, StatusCode};
|
||||
use itertools::Itertools;
|
||||
use quick_cache::sync::Cache;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sqlx::FromRow;
|
||||
@@ -1022,6 +1023,53 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
/// Adds the suffixed form of each tag of an `if_jobs:filter_tags:` scope: in a dev workspace a
|
||||
/// job allowed on `gpu` is queued on `gpu-dev`, and the token must still reach the jobs it starts.
|
||||
fn expand_filter_tags_with_dev_suffix(scopes: &mut [String], suffix: &str) {
|
||||
for scope in scopes.iter_mut() {
|
||||
if let Some(tags) = scope.strip_prefix("if_jobs:filter_tags:") {
|
||||
let expanded = tags
|
||||
.split(',')
|
||||
.flat_map(|t| {
|
||||
let suffixed = windmill_common::worker::with_dev_workspace_suffix(t, suffix);
|
||||
[t.to_string(), suffixed]
|
||||
})
|
||||
.unique()
|
||||
.join(",");
|
||||
*scope = format!("if_jobs:filter_tags:{expanded}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Only while isolation is on and for a dev workspace or a fork, whose jobs get the suffix then:
|
||||
/// otherwise the suffixed tags are other tags the token was not given.
|
||||
async fn expand_filter_tags_for_dev_workspace(
|
||||
db: &DB,
|
||||
workspace_id: &str,
|
||||
scopes: Option<&mut Vec<String>>,
|
||||
) {
|
||||
let Some(scopes) = scopes else { return };
|
||||
if !windmill_common::worker::DEV_WORKSPACE_ISOLATION.load(std::sync::atomic::Ordering::Relaxed)
|
||||
|| !scopes.iter().any(|s| s.starts_with("if_jobs:filter_tags:"))
|
||||
{
|
||||
return;
|
||||
}
|
||||
let is_dev = workspace_id.starts_with(windmill_common::workspaces::WM_FORK_PREFIX)
|
||||
|| sqlx::query_scalar::<_, bool>(
|
||||
"SELECT parent_workspace_id IS NOT NULL FROM workspace WHERE id = $1",
|
||||
)
|
||||
.bind(workspace_id)
|
||||
.fetch_optional(db)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.unwrap_or(false);
|
||||
if is_dev {
|
||||
let suffix = windmill_common::worker::DEV_WORKSPACE_TAG_SUFFIX.load();
|
||||
expand_filter_tags_with_dev_suffix(scopes, &suffix);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn transform_old_scope_to_new_scope(scopes: Option<&mut Vec<String>>) {
|
||||
if let Some(scopes) = scopes {
|
||||
for scope in scopes.iter_mut() {
|
||||
@@ -1172,6 +1220,14 @@ pub async fn resolve_opt_job_authed(
|
||||
let authed = &mut opt_job_authed.authed;
|
||||
if authed.scopes.is_some() && !own_job_runtime_route {
|
||||
transform_old_scope_to_new_scope(authed.scopes.as_mut());
|
||||
if let Some(w_id) = workspace_id.as_deref() {
|
||||
expand_filter_tags_for_dev_workspace(
|
||||
&cache.db,
|
||||
w_id,
|
||||
authed.scopes.as_mut(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
if let Err(err) = crate::scopes::check_scopes_for_route(
|
||||
authed.scopes.as_deref(),
|
||||
@@ -1359,6 +1415,19 @@ pub async fn list_tokens_internal(
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn filter_tags_reach_their_dev_workspace_form() {
|
||||
let mut scopes = vec![
|
||||
"jobs:read".to_string(),
|
||||
"if_jobs:filter_tags:gpu,bash-dev".to_string(),
|
||||
];
|
||||
expand_filter_tags_with_dev_suffix(&mut scopes, "-dev");
|
||||
assert_eq!(
|
||||
scopes,
|
||||
vec!["jobs:read", "if_jobs:filter_tags:gpu,gpu-dev,bash-dev"]
|
||||
);
|
||||
}
|
||||
|
||||
/// Dropping the `AUTH_CACHE` entry between the two resolutions is what makes a memo
|
||||
/// hit observable: only a working memo can answer the second one.
|
||||
#[tokio::test]
|
||||
|
||||
@@ -273,18 +273,22 @@ async fn update_config(
|
||||
}
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
let was_paused = if name.starts_with("worker__") {
|
||||
let previous = if name.starts_with("worker__") {
|
||||
sqlx::query_scalar!(
|
||||
"SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE",
|
||||
"SELECT config FROM config WHERE name = $1 FOR UPDATE",
|
||||
&name
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten()
|
||||
.unwrap_or(false)
|
||||
} else {
|
||||
false
|
||||
None
|
||||
};
|
||||
let is_set = |c: Option<&serde_json::Value>, k: &str| {
|
||||
c.and_then(|c| c.get(k)).and_then(|v| v.as_bool()) == Some(true)
|
||||
};
|
||||
let was_paused = is_set(previous.as_ref(), "paused");
|
||||
let was_serving_dev = is_set(previous.as_ref(), "serves_dev_workspaces");
|
||||
sqlx::query!(
|
||||
"INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config",
|
||||
&name,
|
||||
@@ -306,7 +310,7 @@ async fn update_config(
|
||||
tx.commit().await?;
|
||||
|
||||
if name.starts_with("worker__") {
|
||||
let paused = config.get("paused").and_then(|v| v.as_bool()) == Some(true);
|
||||
let paused = is_set(Some(&config), "paused");
|
||||
if paused != was_paused {
|
||||
windmill_common::feature_usage::log_feature_usage(
|
||||
"worker_group",
|
||||
@@ -314,6 +318,14 @@ async fn update_config(
|
||||
if paused { "pause" } else { "resume" },
|
||||
);
|
||||
}
|
||||
let serving_dev = is_set(Some(&config), "serves_dev_workspaces");
|
||||
if serving_dev != was_serving_dev {
|
||||
windmill_common::feature_usage::log_feature_usage(
|
||||
"worker_group",
|
||||
"dev_workspaces_toggle",
|
||||
if serving_dev { "serve" } else { "stop" },
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(format!("Updated config {name}"))
|
||||
}
|
||||
|
||||
@@ -434,6 +434,7 @@ pub async fn import_completed_jobs(
|
||||
|
||||
pub async fn import_queued_jobs(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<windmill_common::DB>,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path(w_id): Path<String>,
|
||||
Json(jobs): Json<Vec<ExportableQueuedJob>>,
|
||||
@@ -468,7 +469,9 @@ pub async fn import_queued_jobs(
|
||||
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
|
||||
for job in jobs {
|
||||
for mut job in jobs {
|
||||
// A fork's admin is whoever created it, so an imported tag is as untrusted as a pushed one.
|
||||
job.tag = windmill_queue::tags::apply_dev_workspace_suffix(job.tag, &w_id, &db).await?;
|
||||
sqlx::query!(
|
||||
r#"
|
||||
INSERT INTO v2_job (
|
||||
|
||||
@@ -916,6 +916,7 @@ pub async fn set_global_setting_internal(
|
||||
|
||||
run_setting_pre_write_hook(db, &key, &value).await?;
|
||||
|
||||
let turns_on = value.as_bool() == Some(true);
|
||||
match value {
|
||||
serde_json::Value::Null => {
|
||||
if instance_config::PROTECTED_SETTINGS.contains(&key.as_str()) {
|
||||
@@ -949,6 +950,14 @@ pub async fn set_global_setting_internal(
|
||||
}
|
||||
};
|
||||
|
||||
if key == windmill_common::global_settings::DEV_WORKSPACE_ISOLATION_SETTING {
|
||||
windmill_common::feature_usage::log_feature_usage(
|
||||
"dev_workspace_isolation",
|
||||
"toggle",
|
||||
if turns_on { "on" } else { "off" },
|
||||
);
|
||||
}
|
||||
|
||||
if should_bump_instance_ai_revision {
|
||||
bump_instance_ai_config_revision();
|
||||
}
|
||||
@@ -975,6 +984,14 @@ async fn run_setting_pre_write_hook(
|
||||
value: &serde_json::Value,
|
||||
) -> error::Result<()> {
|
||||
match key {
|
||||
windmill_common::global_settings::DEV_WORKSPACE_TAG_SUFFIX_SETTING => match value {
|
||||
serde_json::Value::Null => {}
|
||||
serde_json::Value::String(s) if s.trim().is_empty() => {}
|
||||
suffix => windmill_common::global_settings::validate_dev_workspace_tag_suffix(suffix)
|
||||
.map_err(|e| {
|
||||
error::Error::BadRequest(format!("The dev workspace tag suffix {e}"))
|
||||
})?,
|
||||
},
|
||||
// The instance AI config is written as an untyped blob through this generic
|
||||
// endpoint, so it never passes the typed check the workspace handler applies.
|
||||
// Rates that reach a cost total unbounded would make it negative or infinite.
|
||||
|
||||
@@ -11,6 +11,7 @@ path = "src/lib.rs"
|
||||
[dependencies]
|
||||
windmill-api-auth.workspace = true
|
||||
windmill-common = { workspace = true, default-features = false }
|
||||
windmill-queue.workspace = true
|
||||
axum.workspace = true
|
||||
chrono.workspace = true
|
||||
serde.workspace = true
|
||||
|
||||
@@ -21,7 +21,10 @@ use windmill_common::{
|
||||
jobs::{HIDE_WORKERS_FOR_NON_ADMINS, TAGS_ARE_SENSITIVE},
|
||||
queue_metrics::{read_queue_metrics_series, QueueMetricsSeries},
|
||||
utils::{paginate, Pagination},
|
||||
worker::{ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE},
|
||||
worker::{
|
||||
ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE,
|
||||
DEV_WORKSPACE_ISOLATION, DEV_WORKSPACE_TAG_SUFFIX,
|
||||
},
|
||||
workspaces::workspace_with_fork_ancestors,
|
||||
DB,
|
||||
};
|
||||
@@ -37,6 +40,7 @@ pub fn global_service() -> Router {
|
||||
"/is_default_tags_per_workspace",
|
||||
get(get_default_tags_per_workspace),
|
||||
)
|
||||
.route("/dev_workspace_isolation", get(get_dev_workspace_isolation))
|
||||
.route("/get_default_tags", get(get_default_tags))
|
||||
.route("/queue_metrics", get(get_queue_metrics))
|
||||
.route("/queue_metrics_series", get(get_queue_metrics_series))
|
||||
@@ -186,7 +190,18 @@ async fn exists_workers_with_tags(
|
||||
let chain = workspace_with_fork_ancestors(&db, workspace).await?;
|
||||
let custom_tags = CUSTOM_TAGS_PER_WORKSPACE.load();
|
||||
let allowed_tags = custom_tags.to_string_vec(Some(&chain));
|
||||
tags.retain(|t| allowed_tags.contains(t));
|
||||
// A dev workspace's jobs run on the suffixed form of the tags it may use.
|
||||
let dev_suffix = windmill_queue::tags::dev_workspace_suffix_for(workspace, &db)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
tags.retain(|t| {
|
||||
allowed_tags.contains(t)
|
||||
|| dev_suffix.as_deref().is_some_and(|s| {
|
||||
t.strip_suffix(s)
|
||||
.is_some_and(|b| allowed_tags.iter().any(|a| a == b))
|
||||
})
|
||||
});
|
||||
} else {
|
||||
// No workspace provided and not superadmin - return empty
|
||||
return Ok(Json(std::collections::HashMap::new()));
|
||||
@@ -259,6 +274,19 @@ async fn get_default_tags_per_workspace() -> JsonResult<bool> {
|
||||
))
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct DevWorkspaceIsolation {
|
||||
enabled: bool,
|
||||
suffix: String,
|
||||
}
|
||||
|
||||
async fn get_dev_workspace_isolation() -> JsonResult<DevWorkspaceIsolation> {
|
||||
Ok(Json(DevWorkspaceIsolation {
|
||||
enabled: DEV_WORKSPACE_ISOLATION.load(std::sync::atomic::Ordering::Relaxed),
|
||||
suffix: DEV_WORKSPACE_TAG_SUFFIX.load().to_string(),
|
||||
}))
|
||||
}
|
||||
|
||||
async fn get_default_tags() -> JsonResult<Vec<String>> {
|
||||
Ok(Json(DEFAULT_TAGS.clone()))
|
||||
}
|
||||
|
||||
@@ -868,11 +868,21 @@ pub(crate) async fn change_workspace_id(
|
||||
"Re-pushing jobs for {} enabled schedules",
|
||||
enabled_schedule_paths.len()
|
||||
);
|
||||
for schedule_path in &enabled_schedule_paths {
|
||||
if let Some(schedule) = get_schedule_opt(&mut *tx, &rw.new_id, schedule_path).await? {
|
||||
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
|
||||
windmill_queue::tags::carry_lineage_over_rename(&old_id, &rw.new_id, &db).await?;
|
||||
let new_id = &rw.new_id;
|
||||
let repushed = async {
|
||||
for schedule_path in &enabled_schedule_paths {
|
||||
if let Some(schedule) = get_schedule_opt(&mut *tx, new_id, schedule_path).await? {
|
||||
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
|
||||
}
|
||||
}
|
||||
Ok::<_, Error>(tx)
|
||||
}
|
||||
.await;
|
||||
if repushed.is_err() {
|
||||
windmill_queue::tags::invalidate_fork_parent_cache(new_id);
|
||||
}
|
||||
let mut tx = repushed?;
|
||||
|
||||
// After every workspace_settings write above: fork cleanup locks a settings row before the
|
||||
// registry, so taking the registry first here would deadlock with it.
|
||||
|
||||
@@ -11413,6 +11413,30 @@ paths:
|
||||
schema:
|
||||
type: boolean
|
||||
|
||||
/workers/dev_workspace_isolation:
|
||||
get:
|
||||
summary: get whether the jobs of dev workspaces and forks run on their own worker groups
|
||||
operationId: getDevWorkspaceIsolation
|
||||
tags:
|
||||
- worker
|
||||
responses:
|
||||
"200":
|
||||
description: dev workspace isolation
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
enabled:
|
||||
type: boolean
|
||||
description: whether new jobs of dev workspaces and forks are queued on suffixed tags
|
||||
suffix:
|
||||
type: string
|
||||
description: the suffix of those tags, which worker groups serving dev workspaces listen with
|
||||
required:
|
||||
- enabled
|
||||
- suffix
|
||||
|
||||
/w/{workspace}/workspace_dependencies/create:
|
||||
post:
|
||||
summary: create workspace dependencies
|
||||
|
||||
@@ -9264,6 +9264,7 @@ async fn run_inline_preview_script(
|
||||
return Err(operator_preview_refusal(job_id));
|
||||
}
|
||||
check_scopes(&authed, || format!("jobs:run"))?;
|
||||
windmill_queue::tags::check_inline_run_allowed(&w_id, &db).await?;
|
||||
if let Some(job_id) = job_id {
|
||||
register_potential_assets_on_inline_execution(job_id, &w_id, &preview);
|
||||
}
|
||||
@@ -9383,6 +9384,7 @@ async fn run_inline_script_inner(
|
||||
args: Option<HashMap<String, Box<JsonRawValue>>>,
|
||||
user_db: Option<UserDB>,
|
||||
) -> error::Result<Response> {
|
||||
windmill_queue::tags::check_inline_run_allowed(&w_id, &db).await?;
|
||||
// An inline run executes with the caller's own token, so it cannot honour a script's
|
||||
// `job_token_scopes`: such a script only runs as a job. A path is resolved here, once, and
|
||||
// the version checked is the version run.
|
||||
@@ -10183,17 +10185,22 @@ async fn add_batch_jobs(
|
||||
|
||||
let language = language.unwrap_or(ScriptLang::Deno);
|
||||
|
||||
let tag = if let Some(dedicated_worker) = dedicated_worker {
|
||||
let tag = match (dedicated_worker, batch_info.tag) {
|
||||
// Same rule as `push`: a dedicated worker would run it with its own unscoped token.
|
||||
if dedicated_worker && path.is_some() && job_token_scopes.is_none() {
|
||||
(Some(true), _) if path.is_some() && job_token_scopes.is_none() => {
|
||||
windmill_common::worker::dedicated_worker_tag(&w_id, &path.clone().unwrap())
|
||||
} else {
|
||||
format!("{}", language.as_str())
|
||||
}
|
||||
} else if let Some(tag) = batch_info.tag {
|
||||
tag
|
||||
} else {
|
||||
format!("{}", language.as_str())
|
||||
(None, Some(tag)) => {
|
||||
windmill_queue::tags::apply_dev_workspace_suffix(tag, &w_id, &db).await?
|
||||
}
|
||||
_ => {
|
||||
windmill_queue::tags::apply_dev_workspace_suffix(
|
||||
language.as_str().to_string(),
|
||||
&w_id,
|
||||
&db,
|
||||
)
|
||||
.await?
|
||||
}
|
||||
};
|
||||
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
|
||||
@@ -10,6 +10,8 @@ pub const FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING: &str =
|
||||
"fork_workspace_tag_append_fork_suffix";
|
||||
pub const PREVIEW_TAGS_OVERRIDE_SETTING: &str = "preview_tags_override";
|
||||
pub const DEPENDENCY_JOB_TAG_SETTING: &str = "dependency_job_tag";
|
||||
pub const DEV_WORKSPACE_ISOLATION_SETTING: &str = "dev_workspace_isolation";
|
||||
pub const DEV_WORKSPACE_TAG_SUFFIX_SETTING: &str = "dev_workspace_tag_suffix";
|
||||
pub const BASE_URL_SETTING: &str = "base_url";
|
||||
pub const WS_BASE_URL_SETTING: &str = "ws_base_url";
|
||||
pub const OAUTH_SETTING: &str = "oauths";
|
||||
@@ -315,6 +317,21 @@ pub fn validate_accent_color(value: &serde_json::Value) -> Result<(), String> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Validate a non-empty [`DEV_WORKSPACE_TAG_SUFFIX_SETTING`] value. Every writer must call
|
||||
/// it: a suffix that fails it is ignored at load, so the workers would listen on a suffix
|
||||
/// the setting does not show.
|
||||
pub fn validate_dev_workspace_tag_suffix(value: &serde_json::Value) -> Result<(), String> {
|
||||
let s = value
|
||||
.as_str()
|
||||
.ok_or_else(|| "must be a string".to_string())?;
|
||||
if !crate::worker::is_valid_dev_workspace_tag_suffix(s.trim()) {
|
||||
return Err(
|
||||
"may only contain letters, digits, '-', '_' and '.', up to 32 characters".to_string(),
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The settings every signed-in browser reads on each full page load.
|
||||
#[derive(serde::Serialize, Clone, Debug, Default, PartialEq)]
|
||||
pub struct InstanceUi {
|
||||
|
||||
@@ -263,6 +263,8 @@ pub struct GlobalSettings {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub default_tags_per_workspace: Option<bool>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub dev_workspace_isolation: Option<bool>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub preview_tags_override: Option<bool>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub disable_hub: Option<bool>,
|
||||
@@ -279,6 +281,8 @@ pub struct GlobalSettings {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub dependency_job_tag: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub dev_workspace_tag_suffix: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub ws_base_url: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub github_app_webhook_base_url: Option<String>,
|
||||
@@ -970,6 +974,9 @@ pub struct WorkerGroupConfig {
|
||||
/// Workers of the group pull no new job while this is set; queued jobs wait.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub paused: Option<bool>,
|
||||
/// The group listens to its tags suffixed with the dev workspace tag suffix, and only to them.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub serves_dev_workspaces: Option<bool>,
|
||||
/// Object store this group's dependency cache uses instead of the instance one. Same shape
|
||||
/// as the instance `object_store_cache_config`.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
@@ -1499,6 +1506,14 @@ pub async fn sync_global_settings_declarative(
|
||||
.map_err(|e| anyhow::anyhow!("{accent_key}: {e}"))?,
|
||||
}
|
||||
|
||||
let suffix_key = crate::global_settings::DEV_WORKSPACE_TAG_SUFFIX_SETTING;
|
||||
match desired.get(suffix_key) {
|
||||
None | Some(serde_json::Value::Null) => {}
|
||||
Some(serde_json::Value::String(s)) if s.trim().is_empty() => {}
|
||||
Some(suffix) => crate::global_settings::validate_dev_workspace_tag_suffix(suffix)
|
||||
.map_err(|e| anyhow::anyhow!("{suffix_key}: {e}"))?,
|
||||
}
|
||||
|
||||
// An origin list that cannot be parsed is dropped at boot, leaving the
|
||||
// empty default — which is no restriction at all. Rejecting it here is what
|
||||
// keeps a typo in a ConfigMap from silently widening CORS instance-wide.
|
||||
|
||||
@@ -451,6 +451,13 @@ lazy_static::lazy_static! {
|
||||
pub static ref FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX: AtomicBool = AtomicBool::new(false);
|
||||
pub static ref PREVIEW_TAGS_OVERRIDE: AtomicBool = AtomicBool::new(false);
|
||||
pub static ref DEPENDENCY_JOB_TAG: arc_swap::ArcSwap<Option<String>> = arc_swap::ArcSwap::from_pointee(None);
|
||||
/// Whether the jobs of dev workspaces and forks are queued on suffixed tags.
|
||||
pub static ref DEV_WORKSPACE_ISOLATION: AtomicBool = AtomicBool::new(false);
|
||||
/// The suffix of those tags. Groups serving dev workspaces listen with it whether isolation is
|
||||
/// on or not, so turning it off leaves no job already queued on a suffixed tag unserved.
|
||||
pub static ref DEV_WORKSPACE_TAG_SUFFIX: arc_swap::ArcSwap<String> = arc_swap::ArcSwap::from_pointee(DEFAULT_DEV_WORKSPACE_TAG_SUFFIX.to_string());
|
||||
/// Whether this process's worker group serves dev workspaces, set when its config is loaded.
|
||||
pub static ref WORKER_SERVES_DEV_WORKSPACES: AtomicBool = AtomicBool::new(false);
|
||||
|
||||
pub static ref MAX_TIMEOUT: u64 = std::env::var("TIMEOUT")
|
||||
.ok()
|
||||
@@ -2418,9 +2425,45 @@ pub async fn update_worker_ping_main_loop_query(
|
||||
// occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, vcpus = COALESCE($7, vcpus),
|
||||
// memory = COALESCE($8, memory), occupancy_rate_15s = $9, occupancy_rate_5m = $10, occupancy_rate_30m = $11 WHERE worker = $6",
|
||||
|
||||
const MAX_TAG_LEN: usize = 50;
|
||||
/// The length of `v2_job.tag`.
|
||||
pub const MAX_TAG_LEN: usize = 50;
|
||||
const HASH_SUFFIX_LEN: usize = 16;
|
||||
|
||||
pub const DEFAULT_DEV_WORKSPACE_TAG_SUFFIX: &str = "-dev";
|
||||
|
||||
/// The suffix new jobs of dev workspaces and forks get, `None` while isolation is off.
|
||||
pub fn dev_workspace_tag_suffix() -> Option<String> {
|
||||
DEV_WORKSPACE_ISOLATION
|
||||
.load(std::sync::atomic::Ordering::Relaxed)
|
||||
.then(|| DEV_WORKSPACE_TAG_SUFFIX.load().to_string())
|
||||
}
|
||||
|
||||
/// Worker tags are pasted into the pull queries as SQL literals and listed comma-separated, so
|
||||
/// the suffix must not carry a quote, a comma or whitespace.
|
||||
pub fn is_valid_dev_workspace_tag_suffix(suffix: &str) -> bool {
|
||||
!suffix.is_empty()
|
||||
&& suffix.len() <= 32
|
||||
&& suffix
|
||||
.chars()
|
||||
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
|
||||
}
|
||||
|
||||
/// `tag` as the workers serving dev workspaces listen to it. Idempotent, because a tag inherited
|
||||
/// from a dev job (a flow step, an agent's tool) already carries the suffix.
|
||||
pub fn with_dev_workspace_suffix(tag: &str, suffix: &str) -> String {
|
||||
if tag.ends_with(suffix) {
|
||||
tag.to_string()
|
||||
} else {
|
||||
format!("{tag}{suffix}")
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `tag` has the shape of a [`dedicated_worker_tag`] too long to keep whole. Its prefix
|
||||
/// alone does not tell whose it is: two workspaces may share it.
|
||||
pub fn is_hashed_dedicated_worker_tag(tag: &str) -> bool {
|
||||
tag.len() == MAX_TAG_LEN && tag.as_bytes().get(MAX_TAG_LEN - 1 - HASH_SUFFIX_LEN) == Some(&b'#')
|
||||
}
|
||||
|
||||
pub fn dedicated_worker_tag(workspace_id: &str, path: &str) -> String {
|
||||
let full_tag = format!("{}:{}", workspace_id, path);
|
||||
if full_tag.len() <= MAX_TAG_LEN {
|
||||
@@ -2581,8 +2624,32 @@ pub async fn load_worker_config(
|
||||
})
|
||||
.unwrap_or_else(|| DEFAULT_TAGS.clone());
|
||||
|
||||
// A group serving dev workspaces listens to the suffixed form of its tags only, so it never
|
||||
// takes a root workspace's job. Dedicated workers are placed by hand and keep their tags.
|
||||
let dev_suffix = (config.serves_dev_workspaces == Some(true)
|
||||
&& dedicated_worker.is_none()
|
||||
&& dedicated_workers.is_none())
|
||||
.then(|| DEV_WORKSPACE_TAG_SUFFIX.load().to_string());
|
||||
WORKER_SERVES_DEV_WORKSPACES.store(dev_suffix.is_some(), std::sync::atomic::Ordering::Relaxed);
|
||||
let worker_tags = match &dev_suffix {
|
||||
Some(suffix) => worker_tags
|
||||
.iter()
|
||||
.map(|t| with_dev_workspace_suffix(t, suffix))
|
||||
.unique()
|
||||
.collect_vec(),
|
||||
None => worker_tags,
|
||||
};
|
||||
|
||||
let mut priority_tags_sorted: Vec<PriorityTags> = Vec::new();
|
||||
let priority_tags_map = config.priority_tags.unwrap_or_else(HashMap::new);
|
||||
let priority_tags_map: HashMap<String, u8> = match &dev_suffix {
|
||||
Some(suffix) => config
|
||||
.priority_tags
|
||||
.unwrap_or_default()
|
||||
.into_iter()
|
||||
.map(|(t, p)| (with_dev_workspace_suffix(&t, suffix), p))
|
||||
.collect(),
|
||||
None => config.priority_tags.unwrap_or_else(HashMap::new),
|
||||
};
|
||||
if priority_tags_map.len() > 0 {
|
||||
let mut all_tags_set: HashSet<String> = HashSet::from_iter(worker_tags.clone());
|
||||
|
||||
@@ -2778,6 +2845,7 @@ pub struct WorkerConfigOpt {
|
||||
pub native_mode: Option<bool>,
|
||||
pub object_store_cache_config: Option<serde_json::Value>,
|
||||
pub paused: Option<bool>,
|
||||
pub serves_dev_workspaces: Option<bool>,
|
||||
}
|
||||
|
||||
impl Default for WorkerConfigOpt {
|
||||
@@ -2798,6 +2866,7 @@ impl Default for WorkerConfigOpt {
|
||||
native_mode: Default::default(),
|
||||
object_store_cache_config: Default::default(),
|
||||
paused: Default::default(),
|
||||
serves_dev_workspaces: Default::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4778,6 +4778,9 @@ pub async fn pull(
|
||||
}
|
||||
}
|
||||
};
|
||||
let tag =
|
||||
crate::tags::apply_dev_workspace_suffix(tag, &job.workspace_id, db)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"UPDATE v2_job_queue SET tag = $1, running = false WHERE id = $2",
|
||||
tag,
|
||||
@@ -7372,7 +7375,8 @@ async fn push_inner<'c, 'd>(
|
||||
|
||||
// A dedicated worker runs every job it serves with its own unscoped worker token, so a
|
||||
// job with a restricted token runs on the regular workers of its language instead.
|
||||
let tag = if dedicated_worker.is_some_and(|x| x) && job_token_scopes.is_none() {
|
||||
let on_dedicated_worker = dedicated_worker.is_some_and(|x| x) && job_token_scopes.is_none();
|
||||
let tag = if on_dedicated_worker {
|
||||
let flow_prefix = if job_kind == JobKind::Flow || job_kind == JobKind::FlowDependencies {
|
||||
"flow/"
|
||||
} else {
|
||||
@@ -7453,6 +7457,11 @@ async fn push_inner<'c, 'd>(
|
||||
}
|
||||
})
|
||||
};
|
||||
let tag = if on_dedicated_worker {
|
||||
tag
|
||||
} else {
|
||||
crate::tags::apply_dev_workspace_suffix(tag, workspace_id, db).await?
|
||||
};
|
||||
|
||||
let mut tx = tx.into_tx().await?;
|
||||
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
use sqlx::{Pool, Postgres};
|
||||
use windmill_common::error::Error;
|
||||
use windmill_common::worker::{
|
||||
DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX,
|
||||
dedicated_worker_tag, dev_workspace_tag_suffix, is_hashed_dedicated_worker_tag,
|
||||
with_dev_workspace_suffix, DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES,
|
||||
DEV_WORKSPACE_TAG_SUFFIX, FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX, MAX_TAG_LEN,
|
||||
WORKER_SERVES_DEV_WORKSPACES,
|
||||
};
|
||||
use windmill_common::workspaces::WM_FORK_PREFIX;
|
||||
|
||||
@@ -13,6 +17,10 @@ struct TagWorkspace {
|
||||
/// Whether the `-fork` suffix may apply: the workspace shares an ancestor's id, or carries the
|
||||
/// generated prefix itself.
|
||||
is_fork: bool,
|
||||
/// Whether the workspace's jobs run on the workers serving dev workspaces: it has a parent,
|
||||
/// or carries the generated prefix (a fork orphaned by its parent's deletion stays a fork).
|
||||
/// `None` when the lookup failed.
|
||||
is_dev: Option<bool>,
|
||||
}
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
@@ -140,9 +148,153 @@ pub async fn per_workspace_tag(workspace_id: &str, db: &Pool<Postgres>) -> Optio
|
||||
})
|
||||
}
|
||||
|
||||
/// `tag`, fully resolved, as a job of `workspace_id` is queued on it: suffixed with the dev
|
||||
/// workspace tag suffix when isolation is on and the workspace is a dev workspace or a fork, so
|
||||
/// only the worker groups serving dev workspaces pick it up. A tag no group serves leaves the job
|
||||
/// queued; it never falls back to the unsuffixed tag. Dedicated worker tags are exempt: they
|
||||
/// already name the workspace and an admin places each dedicated worker by hand.
|
||||
pub async fn apply_dev_workspace_suffix(
|
||||
tag: String,
|
||||
workspace_id: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<String, Error> {
|
||||
let Some(suffix) = dev_workspace_suffix_for(workspace_id, db).await? else {
|
||||
return Ok(tag);
|
||||
};
|
||||
if is_dedicated_worker_tag_of(workspace_id, &tag, db).await? {
|
||||
return Ok(tag);
|
||||
}
|
||||
let suffixed = with_dev_workspace_suffix(&tag, &suffix);
|
||||
if suffixed.len() > MAX_TAG_LEN {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"tag {tag} of a dev workspace job would be {suffixed} once suffixed with {suffix}, \
|
||||
longer than the {MAX_TAG_LEN} characters a job tag holds"
|
||||
)));
|
||||
}
|
||||
Ok(suffixed)
|
||||
}
|
||||
|
||||
/// Whether `tag` is a dedicated worker tag of `workspace_id`. A hashed one keeps only a prefix of
|
||||
/// the workspace id, which another workspace may share, so it counts only when a worker group's
|
||||
/// dedicated config for this workspace produces it.
|
||||
async fn is_dedicated_worker_tag_of(
|
||||
workspace_id: &str,
|
||||
tag: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<bool, Error> {
|
||||
if tag.starts_with(&format!("{workspace_id}:")) {
|
||||
return Ok(true);
|
||||
}
|
||||
if !is_hashed_dedicated_worker_tag(tag) {
|
||||
return Ok(false);
|
||||
}
|
||||
let configs = sqlx::query_scalar!("SELECT config FROM config WHERE name LIKE 'worker\\_\\_%'")
|
||||
.fetch_all(db)
|
||||
.await?;
|
||||
Ok(configs.into_iter().flatten().any(|config| {
|
||||
let single = config.get("dedicated_worker").cloned().into_iter();
|
||||
let many = config
|
||||
.get("dedicated_workers")
|
||||
.and_then(|v| v.as_array().cloned())
|
||||
.unwrap_or_default();
|
||||
single.chain(many).any(|entry| {
|
||||
entry
|
||||
.as_str()
|
||||
.and_then(|e| e.split_once(':'))
|
||||
.is_some_and(|(ws, path)| {
|
||||
ws == workspace_id && dedicated_worker_tag(ws, path) == tag
|
||||
})
|
||||
})
|
||||
}))
|
||||
}
|
||||
|
||||
/// Gives `new_id` the lineage of `old_id` before a rename commits: the rename re-pushes the
|
||||
/// workspace's schedules in its own transaction, where the new row is not yet visible to the
|
||||
/// lookup, and a job queued then would keep a tag resolved without its parent for good.
|
||||
pub async fn carry_lineage_over_rename(
|
||||
old_id: &str,
|
||||
new_id: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<(), Error> {
|
||||
let old = lookup_tag_workspace(old_id, db).await;
|
||||
if old.is_dev.is_none() {
|
||||
return Err(Error::internal_err(format!(
|
||||
"could not resolve the lineage of {old_id}"
|
||||
)));
|
||||
}
|
||||
let new = TagWorkspace {
|
||||
id: if old.id == old_id {
|
||||
new_id.to_string()
|
||||
} else {
|
||||
old.id
|
||||
},
|
||||
..old
|
||||
};
|
||||
FORK_PARENT_CACHE.insert(new_id.to_string(), (new, std::time::Instant::now()));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The suffix to strip from a flow job's tag to compare it with the generic flow tags: that of a
|
||||
/// dev workspace's flow, isolation on or off, so a flow started before isolation was turned off
|
||||
/// still routes its steps by their own tags.
|
||||
pub async fn dev_flow_tag_suffix(
|
||||
workspace_id: &str,
|
||||
flow_tag: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<Option<String>, Error> {
|
||||
let suffix = DEV_WORKSPACE_TAG_SUFFIX.load().to_string();
|
||||
if !flow_tag.ends_with(&suffix) {
|
||||
return Ok(None);
|
||||
}
|
||||
match lookup_tag_workspace(workspace_id, db).await.is_dev {
|
||||
Some(true) => Ok(Some(suffix)),
|
||||
Some(false) => Ok(None),
|
||||
None => Err(Error::internal_err(format!(
|
||||
"could not resolve whether {workspace_id} is a dev workspace"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// The suffix the jobs of `workspace_id` are queued with: `Some` while isolation is on and the
|
||||
/// workspace is a dev workspace or a fork. A lineage that cannot be read is an error rather than
|
||||
/// a guess, as either guess queues a job on the wrong pool for good.
|
||||
pub async fn dev_workspace_suffix_for(
|
||||
workspace_id: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<Option<String>, Error> {
|
||||
let Some(suffix) = dev_workspace_tag_suffix() else {
|
||||
return Ok(None);
|
||||
};
|
||||
match lookup_tag_workspace(workspace_id, db).await.is_dev {
|
||||
Some(true) => Ok(Some(suffix)),
|
||||
Some(false) => Ok(None),
|
||||
None => Err(Error::internal_err(format!(
|
||||
"could not resolve whether {workspace_id} is a dev workspace, which decides the \
|
||||
workers its jobs run on"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Refuses running code of `workspace_id` in this process, as the inline routes do, when its
|
||||
/// jobs are isolated on dev workers and this process's group does not serve dev workspaces.
|
||||
pub async fn check_inline_run_allowed(
|
||||
workspace_id: &str,
|
||||
db: &Pool<Postgres>,
|
||||
) -> Result<(), Error> {
|
||||
if dev_workspace_suffix_for(workspace_id, db).await?.is_some()
|
||||
&& !WORKER_SERVES_DEV_WORKSPACES.load(std::sync::atomic::Ordering::Relaxed)
|
||||
{
|
||||
return Err(Error::BadRequest(format!(
|
||||
"{workspace_id} is a dev workspace or a fork, whose code runs only on worker groups \
|
||||
serving dev workspaces, and this worker's group does not: run it as a job instead"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Backed by a short-TTL cache to avoid a DB round-trip per job push. A transient DB error resolves
|
||||
/// to the workspace's own id for this call but is NOT cached, so the next push retries instead of
|
||||
/// misrouting for the whole TTL.
|
||||
/// misrouting for the whole TTL. Whether the workspace is dev is then unknown.
|
||||
async fn lookup_tag_workspace(workspace_id: &str, db: &Pool<Postgres>) -> TagWorkspace {
|
||||
if let Some((resolved, cached_at)) = FORK_PARENT_CACHE.get(workspace_id) {
|
||||
if cached_at.elapsed().as_secs() < FORK_PARENT_CACHE_TTL_SECS {
|
||||
@@ -154,7 +306,7 @@ async fn lookup_tag_workspace(workspace_id: &str, db: &Pool<Postgres>) -> TagWor
|
||||
// itself under a generated fork id. The depth bound mirrors the other chain walkers as a
|
||||
// cycle-safety backstop. A `wm-fork-` workspace orphaned by `ON DELETE SET NULL` has no parent
|
||||
// left to borrow from, so it ends the walk on itself.
|
||||
let resolved = match sqlx::query_scalar!(
|
||||
let resolved = match sqlx::query!(
|
||||
r#"
|
||||
WITH RECURSIVE chain AS (
|
||||
SELECT id, parent_workspace_id, is_dev_workspace, 0 AS depth
|
||||
@@ -165,35 +317,47 @@ async fn lookup_tag_workspace(workspace_id: &str, db: &Pool<Postgres>) -> TagWor
|
||||
JOIN chain ON w.id = chain.parent_workspace_id
|
||||
WHERE chain.depth < 20
|
||||
)
|
||||
SELECT id AS "id!" FROM chain
|
||||
WHERE parent_workspace_id IS NULL
|
||||
OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%')
|
||||
ORDER BY depth LIMIT 1
|
||||
SELECT
|
||||
(SELECT id FROM chain
|
||||
WHERE parent_workspace_id IS NULL
|
||||
OR (is_dev_workspace AND id NOT LIKE 'wm-fork-%')
|
||||
ORDER BY depth LIMIT 1) AS id,
|
||||
(SELECT parent_workspace_id IS NOT NULL FROM chain WHERE depth = 0) AS has_parent
|
||||
"#,
|
||||
workspace_id
|
||||
)
|
||||
.fetch_optional(db)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
{
|
||||
// No row means the workspace does not exist yet, or its chain is broken: both are "we
|
||||
// cannot resolve this", so fall back to its own id for this call WITHOUT caching. Caching
|
||||
// it would pin a fork to its own unserved id for the whole TTL — a rename resolves the new
|
||||
// id before the row lands, and every job pushed until the TTL expires would queue.
|
||||
Ok(None) => {
|
||||
return TagWorkspace {
|
||||
id: workspace_id.to_string(),
|
||||
is_fork: workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
Ok(row) => match row.id {
|
||||
Some(id) => TagWorkspace {
|
||||
is_fork: id != workspace_id || workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
is_dev: Some(
|
||||
row.has_parent == Some(true) || workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
),
|
||||
id,
|
||||
},
|
||||
// No id means the workspace does not exist yet, or its chain is broken: both are "we
|
||||
// cannot resolve this", so fall back to its own id for this call WITHOUT caching.
|
||||
// Caching it would pin a fork to its own unserved id for the whole TTL — a rename
|
||||
// resolves the new id before the row lands, and every job pushed until the TTL
|
||||
// expires would queue.
|
||||
None => {
|
||||
return TagWorkspace {
|
||||
id: workspace_id.to_string(),
|
||||
is_fork: workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
is_dev: Some(
|
||||
row.has_parent == Some(true) || workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Some(id)) => {
|
||||
let is_fork = id != workspace_id || workspace_id.starts_with(WM_FORK_PREFIX);
|
||||
TagWorkspace { id, is_fork }
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
tracing::warn!("failed to resolve tag workspace for {workspace_id}: {e:#}");
|
||||
return TagWorkspace {
|
||||
id: workspace_id.to_string(),
|
||||
is_fork: workspace_id.starts_with(WM_FORK_PREFIX),
|
||||
is_dev: None,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
@@ -577,6 +577,9 @@ async fn enqueue_windmill_tool(
|
||||
&ctx.job.workspace_id,
|
||||
ctx.preserve_step_tags,
|
||||
job_payload.tag.as_deref(),
|
||||
windmill_queue::tags::dev_flow_tag_suffix(&ctx.job.workspace_id, &ctx.job.tag, ctx.db)
|
||||
.await?
|
||||
.as_deref(),
|
||||
);
|
||||
if tool_tag.as_deref().is_some_and(tag_reads_flow_expr) {
|
||||
append_logs(
|
||||
|
||||
@@ -1631,7 +1631,17 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
Some(t) => {
|
||||
let no_args = HashMap::new();
|
||||
let args = PushArgs::from(fetched_args.as_ref().unwrap_or(&no_args));
|
||||
resolve_push_tag(t, &args, &flow_job.workspace_id, db).await
|
||||
match resolve_push_tag(t, &args, &flow_job.workspace_id, db).await {
|
||||
Some(t) => Some(
|
||||
windmill_queue::tags::apply_dev_workspace_suffix(
|
||||
t,
|
||||
&flow_job.workspace_id,
|
||||
db,
|
||||
)
|
||||
.await?,
|
||||
),
|
||||
None => None,
|
||||
}
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
@@ -2210,7 +2220,8 @@ pub async fn update_flow_status_after_job_completion_internal(
|
||||
None
|
||||
},
|
||||
has_triggered_error_handler: has_triggered_error_handler || is_failure_step,
|
||||
recovered_by_child: failure_step_recovers || (recovered_by_child && !success),
|
||||
recovered_by_child: failure_step_recovers
|
||||
|| (recovered_by_child && !success),
|
||||
},
|
||||
));
|
||||
}
|
||||
@@ -3284,8 +3295,17 @@ pub(crate) fn resolve_flow_step_tag(
|
||||
workspace_id: &str,
|
||||
preserve_step_tags: bool,
|
||||
step_tag: Option<&str>,
|
||||
// The suffix of a dev workspace's flow tag, if any: its flows run on the generic tag suffixed,
|
||||
// and their steps are suffixed when pushed.
|
||||
dev_suffix: Option<&str>,
|
||||
) -> Option<String> {
|
||||
if is_preprocessor_step || flow_tag == "flow" || flow_tag == format!("flow-{}", workspace_id) {
|
||||
let generic_flow_tag = dev_suffix
|
||||
.and_then(|s| flow_tag.strip_suffix(s))
|
||||
.unwrap_or(flow_tag);
|
||||
if is_preprocessor_step
|
||||
|| generic_flow_tag == "flow"
|
||||
|| generic_flow_tag == format!("flow-{}", workspace_id)
|
||||
{
|
||||
step_tag.map(str::to_string)
|
||||
} else if preserve_step_tags && step_tag.is_some_and(|t| !t.is_empty()) {
|
||||
step_tag.map(str::to_string)
|
||||
@@ -3363,7 +3383,14 @@ mod tag_resolution_tests {
|
||||
// Parent flow on a custom tag, step declares its own tag, preserve disabled:
|
||||
// the step inherits the flow tag (historical behavior).
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", false, Some("worker-group-B")),
|
||||
resolve_flow_step_tag(
|
||||
false,
|
||||
"worker-group-A",
|
||||
"w1",
|
||||
false,
|
||||
Some("worker-group-B"),
|
||||
None
|
||||
),
|
||||
Some("worker-group-A".to_string())
|
||||
);
|
||||
}
|
||||
@@ -3373,7 +3400,14 @@ mod tag_resolution_tests {
|
||||
// The exact customer scenario: a sub-flow tagged worker-group-B run as a step of a
|
||||
// flow tagged worker-group-A now runs on worker-group-B when preserve_step_tags is on.
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("worker-group-B")),
|
||||
resolve_flow_step_tag(
|
||||
false,
|
||||
"worker-group-A",
|
||||
"w1",
|
||||
true,
|
||||
Some("worker-group-B"),
|
||||
None
|
||||
),
|
||||
Some("worker-group-B".to_string())
|
||||
);
|
||||
}
|
||||
@@ -3381,12 +3415,12 @@ mod tag_resolution_tests {
|
||||
#[test]
|
||||
fn untagged_step_inherits_flow_tag_even_when_preserve_enabled() {
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", true, None),
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", true, None, None),
|
||||
Some("worker-group-A".to_string())
|
||||
);
|
||||
// An empty tag counts as "no tag" and still inherits.
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some("")),
|
||||
resolve_flow_step_tag(false, "worker-group-A", "w1", true, Some(""), None),
|
||||
Some("worker-group-A".to_string())
|
||||
);
|
||||
}
|
||||
@@ -3395,16 +3429,38 @@ mod tag_resolution_tests {
|
||||
fn generic_flow_tag_always_uses_step_tag() {
|
||||
for flow_tag in ["flow", "flow-w1"] {
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, flow_tag, "w1", false, Some("worker-group-B")),
|
||||
resolve_flow_step_tag(false, flow_tag, "w1", false, Some("worker-group-B"), None),
|
||||
Some("worker-group-B".to_string())
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dev_workspace_flow_on_suffixed_generic_tag_uses_step_tag() {
|
||||
for flow_tag in ["flow-dev", "flow-w1-dev"] {
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, flow_tag, "w1", false, Some("gpu"), Some("-dev")),
|
||||
Some("gpu".to_string())
|
||||
);
|
||||
}
|
||||
// In a root workspace `flow-dev` is a custom tag the steps inherit.
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(false, "flow-dev", "w1", false, Some("gpu"), None),
|
||||
Some("flow-dev".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preprocessor_step_uses_step_tag() {
|
||||
assert_eq!(
|
||||
resolve_flow_step_tag(true, "worker-group-A", "w1", false, Some("worker-group-B")),
|
||||
resolve_flow_step_tag(
|
||||
true,
|
||||
"worker-group-A",
|
||||
"w1",
|
||||
false,
|
||||
Some("worker-group-B"),
|
||||
None
|
||||
),
|
||||
Some("worker-group-B".to_string())
|
||||
);
|
||||
}
|
||||
@@ -4654,6 +4710,9 @@ async fn push_next_flow_job(
|
||||
&flow_job.workspace_id,
|
||||
flow.preserve_step_tags,
|
||||
payload_tag.tag.as_deref(),
|
||||
windmill_queue::tags::dev_flow_tag_suffix(&flow_job.workspace_id, &flow_job.tag, db)
|
||||
.await?
|
||||
.as_deref(),
|
||||
);
|
||||
|
||||
// A step whose inputs failed to evaluate, or whose `$flow_expr[...]` tag failed to resolve,
|
||||
@@ -4794,10 +4853,8 @@ async fn push_next_flow_job(
|
||||
// narrows that further.
|
||||
windmill_common::scopes::intersect_job_token_scopes(
|
||||
flow_job.job_token_scopes.as_deref(),
|
||||
windmill_common::scopes::step_job_token_scopes(
|
||||
module.job_token_scopes.as_deref(),
|
||||
)
|
||||
.as_deref(),
|
||||
windmill_common::scopes::step_job_token_scopes(module.job_token_scopes.as_deref())
|
||||
.as_deref(),
|
||||
)
|
||||
.as_deref(),
|
||||
)
|
||||
|
||||
@@ -19,6 +19,11 @@
|
||||
import Section from './Section.svelte'
|
||||
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
|
||||
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
|
||||
import {
|
||||
devWorkspaceTagSuffix,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
interface Props {
|
||||
defaultTagPerWorkspace?: boolean | undefined
|
||||
defaultTagWorkspaces?: string[]
|
||||
@@ -134,6 +139,7 @@
|
||||
|
||||
loadDefaultTags()
|
||||
loadWorkspaces()
|
||||
loadDevWorkspaceTagSuffix()
|
||||
</script>
|
||||
|
||||
<Section label="Default tags">
|
||||
@@ -208,6 +214,22 @@
|
||||
/>
|
||||
{/snippet}
|
||||
</ToggleButtonGroup>
|
||||
{#if $devWorkspaceTagSuffix}
|
||||
<span class="text-2xs text-secondary">
|
||||
Dev workspace isolation (Workers page) is on and applies after this: a fork's
|
||||
<code
|
||||
>{forkAppendForkSuffix ? 'python3-{parent_id}-fork' : 'python3-{parent_id}'}</code
|
||||
>
|
||||
runs as
|
||||
<code
|
||||
>{withDevSuffix(
|
||||
forkAppendForkSuffix ? 'python3-{parent_id}-fork' : 'python3-{parent_id}',
|
||||
$devWorkspaceTagSuffix
|
||||
)}</code
|
||||
>, on the worker groups serving dev workspaces. To split root and dev jobs across
|
||||
worker pools, isolation alone is enough.
|
||||
</span>
|
||||
{/if}
|
||||
</div>
|
||||
<Toggle
|
||||
bind:checked={limitToWorkspaces}
|
||||
|
||||
@@ -0,0 +1,495 @@
|
||||
<script lang="ts">
|
||||
import { Alert, Button, Section } from './common'
|
||||
import ConfirmationModal from './common/confirmationModal/ConfirmationModal.svelte'
|
||||
import Toggle from './Toggle.svelte'
|
||||
import TextInput from './text_input/TextInput.svelte'
|
||||
import Select from './select/Select.svelte'
|
||||
import Tooltip from './meltComponents/Tooltip.svelte'
|
||||
import { ArrowRight, ChevronDown, ChevronRight, Save, Settings } from 'lucide-svelte'
|
||||
import {
|
||||
SettingService,
|
||||
WorkerService,
|
||||
WorkspaceService,
|
||||
type GetQueueStatusResponse,
|
||||
type Workspace
|
||||
} from '$lib/gen'
|
||||
import { DEV_WORKSPACE_ISOLATION_SETTING, DEV_WORKSPACE_TAG_SUFFIX_SETTING } from '$lib/consts'
|
||||
import {
|
||||
DEFAULT_DEV_WORKSPACE_TAG_SUFFIX,
|
||||
DEV_WORKSPACE_TAG_SUFFIX_PATTERN,
|
||||
devWorkspaceIsolation,
|
||||
devWorkspaceListenSuffix,
|
||||
devWorkspaceTagSuffix,
|
||||
isDevOrForkWorkspace,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
import { enterpriseLicense } from '$lib/stores'
|
||||
import { sendUserToast } from '$lib/toast'
|
||||
import { pluralize } from '$lib/utils'
|
||||
import { onMount } from 'svelte'
|
||||
|
||||
interface Props {
|
||||
/** Worker group configs, by group name. */
|
||||
workerGroups: Record<string, any> | undefined
|
||||
/** Workers of each group that pinged in the last 15 seconds. */
|
||||
liveWorkers: Record<string, number>
|
||||
canEdit: boolean
|
||||
onConfigureGroup: (name: string) => void
|
||||
}
|
||||
|
||||
let { workerGroups, liveWorkers, canEdit, onConfigureGroup }: Props = $props()
|
||||
|
||||
let suffix = $derived($devWorkspaceTagSuffix)
|
||||
let enabled = $derived(suffix != null)
|
||||
let effectiveSuffix = $derived($devWorkspaceListenSuffix)
|
||||
|
||||
// Overridden while a confirmation is pending, reset to the saved state otherwise.
|
||||
let toggleChecked = $derived(enabled)
|
||||
let suffixDraft = $derived(effectiveSuffix)
|
||||
let suffixError = $derived(
|
||||
DEV_WORKSPACE_TAG_SUFFIX_PATTERN.test(suffixDraft.trim())
|
||||
? undefined
|
||||
: "Use 1 to 32 letters, digits, '-', '_' or '.'"
|
||||
)
|
||||
|
||||
let workspaces = $state<Workspace[] | undefined>(undefined)
|
||||
let devWorkspaces = $derived(
|
||||
(workspaces ?? []).filter((w) => !w.deleted && isDevOrForkWorkspace(w))
|
||||
)
|
||||
let workspaceNames = $derived(Object.fromEntries((workspaces ?? []).map((w) => [w.id, w.name])))
|
||||
|
||||
async function loadWorkspaces() {
|
||||
const all: Workspace[] = []
|
||||
for (let page = 1; ; page++) {
|
||||
const batch = await WorkspaceService.listWorkspacesAsSuperAdmin({ page, perPage: 100 })
|
||||
all.push(...batch)
|
||||
if (batch.length < 100) break
|
||||
}
|
||||
workspaces = all
|
||||
}
|
||||
|
||||
let queueStatus = $state<GetQueueStatusResponse | undefined>(undefined)
|
||||
async function loadQueueStatus() {
|
||||
try {
|
||||
queueStatus = await WorkerService.getQueueStatus()
|
||||
} catch {
|
||||
queueStatus = undefined
|
||||
}
|
||||
}
|
||||
|
||||
onMount(() => {
|
||||
loadDevWorkspaceTagSuffix(true)
|
||||
loadWorkspaces().catch((e) => console.error('Could not list workspaces', e))
|
||||
loadQueueStatus()
|
||||
const interval = setInterval(loadQueueStatus, 15_000)
|
||||
return () => clearInterval(interval)
|
||||
})
|
||||
|
||||
type GroupEntry = { name: string; live: number; tags: string[] }
|
||||
|
||||
function isDedicated(config: any): boolean {
|
||||
return config?.dedicated_worker != undefined || (config?.dedicated_workers?.length ?? 0) > 0
|
||||
}
|
||||
|
||||
let flaggedGroups: GroupEntry[] = $derived(
|
||||
Object.entries(workerGroups ?? {})
|
||||
.filter(([_, c]) => c?.serves_dev_workspaces === true && !isDedicated(c))
|
||||
.map(([name, c]) => ({ name, live: liveWorkers[name] ?? 0, tags: c?.worker_tags ?? [] }))
|
||||
)
|
||||
let liveFlaggedWorkers = $derived(flaggedGroups.reduce((acc, g) => acc + g.live, 0))
|
||||
let unflaggedGroupNames = $derived(
|
||||
Object.entries(workerGroups ?? {})
|
||||
.filter(([_, c]) => c?.serves_dev_workspaces !== true && !isDedicated(c))
|
||||
.map(([name]) => name)
|
||||
)
|
||||
// The only way to break isolation: an unflagged group listing a suffixed tag takes dev jobs.
|
||||
let leakingGroups = $derived(
|
||||
Object.entries(workerGroups ?? {})
|
||||
.filter(([_, c]) => c?.serves_dev_workspaces !== true)
|
||||
.map(([name, c]) => ({
|
||||
name,
|
||||
tags: ((c?.worker_tags ?? []) as string[]).filter((t) => t.endsWith(effectiveSuffix))
|
||||
}))
|
||||
.filter((g) => g.tags.length > 0)
|
||||
)
|
||||
let unservedDevTags = $derived(
|
||||
enabled
|
||||
? (queueStatus ?? []).filter(
|
||||
(s) => s.tag.endsWith(effectiveSuffix) && s.waiting > 0 && s.workers === 0
|
||||
)
|
||||
: []
|
||||
)
|
||||
|
||||
let groupToFlag = $state<string | undefined>(undefined)
|
||||
let collapsed = $state(true)
|
||||
let showWorkspaces = $state(false)
|
||||
let showAdvanced = $state(false)
|
||||
|
||||
let confirmEnable = $state(false)
|
||||
let confirmDisable = $state(false)
|
||||
let saving = $state(false)
|
||||
|
||||
async function setIsolation(on: boolean) {
|
||||
saving = true
|
||||
try {
|
||||
await SettingService.setGlobal({
|
||||
key: DEV_WORKSPACE_ISOLATION_SETTING,
|
||||
requestBody: { value: on }
|
||||
})
|
||||
devWorkspaceIsolation.update((i) => ({ suffix: i?.suffix ?? effectiveSuffix, enabled: on }))
|
||||
sendUserToast(
|
||||
on
|
||||
? `Dev workspace and fork jobs now run on tags suffixed with ${effectiveSuffix}`
|
||||
: 'Dev workspace isolation turned off'
|
||||
)
|
||||
} catch (e) {
|
||||
sendUserToast(`Could not save dev workspace isolation: ${e}`, true)
|
||||
toggleChecked = enabled
|
||||
} finally {
|
||||
saving = false
|
||||
}
|
||||
}
|
||||
|
||||
async function saveSuffix(value: string) {
|
||||
saving = true
|
||||
try {
|
||||
await SettingService.setGlobal({
|
||||
key: DEV_WORKSPACE_TAG_SUFFIX_SETTING,
|
||||
requestBody: { value }
|
||||
})
|
||||
devWorkspaceIsolation.update((i) => ({ enabled: i?.enabled ?? false, suffix: value }))
|
||||
sendUserToast(`Dev workspace tag suffix set to ${value}`)
|
||||
} catch (e) {
|
||||
sendUserToast(`Could not save the dev workspace tag suffix: ${e}`, true)
|
||||
} finally {
|
||||
saving = false
|
||||
}
|
||||
}
|
||||
|
||||
function onToggle(checked: boolean) {
|
||||
if (checked === enabled) return
|
||||
collapsed = false
|
||||
if (checked) {
|
||||
if (liveFlaggedWorkers === 0) {
|
||||
confirmEnable = true
|
||||
} else {
|
||||
setIsolation(true)
|
||||
}
|
||||
} else {
|
||||
confirmDisable = true
|
||||
}
|
||||
}
|
||||
|
||||
function workspaceKind(w: Workspace): string {
|
||||
if (w.parent_workspace_id) {
|
||||
const parent = workspaceNames[w.parent_workspace_id] ?? w.parent_workspace_id
|
||||
return w.is_dev_workspace ? `dev workspace of ${parent}` : `fork of ${parent}`
|
||||
}
|
||||
return 'fork whose parent was deleted'
|
||||
}
|
||||
</script>
|
||||
|
||||
<ConfirmationModal
|
||||
open={confirmEnable}
|
||||
title="Turn on dev workspace isolation"
|
||||
confirmationText="Turn on"
|
||||
type="info"
|
||||
onCanceled={() => {
|
||||
confirmEnable = false
|
||||
toggleChecked = enabled
|
||||
}}
|
||||
onConfirmed={async () => {
|
||||
confirmEnable = false
|
||||
await setIsolation(true)
|
||||
}}
|
||||
>
|
||||
<div class="flex flex-col gap-2 text-sm">
|
||||
<p>
|
||||
{#if workspaces === undefined}
|
||||
Every dev workspace and fork
|
||||
{:else}
|
||||
<b>{pluralize(devWorkspaces.length, 'dev workspace or fork', 'dev workspaces and forks')}</b
|
||||
>
|
||||
{/if}
|
||||
will have their jobs queued until a worker group serves them.
|
||||
</p>
|
||||
<p class="text-secondary text-xs">
|
||||
{#if flaggedGroups.length === 0}
|
||||
No worker group is flagged as serving dev workspaces yet.
|
||||
{:else}
|
||||
The worker groups serving dev workspaces ({flaggedGroups.map((g) => g.name).join(', ')})
|
||||
have no live worker.
|
||||
{/if}
|
||||
Jobs never fall back to the other worker groups: they wait in the queue and run as soon as a serving
|
||||
worker is up.
|
||||
</p>
|
||||
{#if unflaggedGroupNames.length > 0 && flaggedGroups.length === 0}
|
||||
<div>
|
||||
<Button
|
||||
variant="default"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: Settings }}
|
||||
onClick={() => {
|
||||
confirmEnable = false
|
||||
toggleChecked = enabled
|
||||
onConfigureGroup(unflaggedGroupNames[0])
|
||||
}}
|
||||
>
|
||||
Flag a worker group first
|
||||
</Button>
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
</ConfirmationModal>
|
||||
|
||||
<ConfirmationModal
|
||||
open={confirmDisable}
|
||||
title="Turn off dev workspace isolation"
|
||||
confirmationText="Turn off"
|
||||
type="danger"
|
||||
onCanceled={() => {
|
||||
confirmDisable = false
|
||||
toggleChecked = enabled
|
||||
}}
|
||||
onConfirmed={async () => {
|
||||
confirmDisable = false
|
||||
await setIsolation(false)
|
||||
}}
|
||||
>
|
||||
<p class="text-sm">
|
||||
New jobs of dev workspaces and forks will run on the same worker groups as root workspaces
|
||||
again. Jobs already queued on a <code>{effectiveSuffix}</code> tag keep it, and the worker groups
|
||||
serving dev workspaces still run them.
|
||||
</p>
|
||||
</ConfirmationModal>
|
||||
|
||||
{#snippet isolationToggle()}
|
||||
<Toggle
|
||||
bind:checked={toggleChecked}
|
||||
on:change={(e) => onToggle(e.detail)}
|
||||
disabled={!canEdit || !$enterpriseLicense || saving || suffix === undefined}
|
||||
options={{ right: 'Run dev workspaces and forks on separate workers' }}
|
||||
size="xs"
|
||||
/>
|
||||
{/snippet}
|
||||
|
||||
<Section
|
||||
label="Dev workspace isolation"
|
||||
tooltip="Dev workspaces and forks run code that has not been reviewed or deployed to a root workspace. Route their jobs to worker groups of their own, so they never run on the workers of your root workspaces."
|
||||
eeOnly={!$enterpriseLicense}
|
||||
collapsable
|
||||
bind:collapsed
|
||||
wrapperClass="mt-4 mb-6 rounded-md border border-light p-4 bg-surface-tertiary"
|
||||
>
|
||||
{#snippet action()}
|
||||
{@render isolationToggle()}
|
||||
{/snippet}
|
||||
{#snippet badge()}
|
||||
{@render isolationToggle()}
|
||||
{/snippet}
|
||||
|
||||
<div class="flex flex-col gap-4">
|
||||
<div class="grid grid-cols-[auto_auto_1fr] gap-x-3 gap-y-2 items-center text-xs max-w-2xl">
|
||||
<span class="font-semibold text-emphasis">Root workspaces</span>
|
||||
<ArrowRight size={14} class="text-secondary" />
|
||||
<span class="text-primary">
|
||||
Worker groups not serving dev workspaces, on their usual tags:
|
||||
<code>python3</code>, <code>gpu</code>
|
||||
</span>
|
||||
<span class="font-semibold text-emphasis">Dev workspaces and forks</span>
|
||||
<ArrowRight size={14} class="text-secondary" />
|
||||
<span class="text-primary">
|
||||
{#if enabled}
|
||||
Worker groups serving dev workspaces, on tags suffixed with
|
||||
<code>{effectiveSuffix}</code>:
|
||||
{:else}
|
||||
Once on, worker groups serving dev workspaces, on suffixed tags:
|
||||
{/if}
|
||||
<code>{withDevSuffix('python3', effectiveSuffix)}</code>,
|
||||
<code>{withDevSuffix('gpu', effectiveSuffix)}</code>
|
||||
</span>
|
||||
</div>
|
||||
<p class="text-2xs text-secondary max-w-2xl">
|
||||
A workspace with a parent, and every fork (including a fork whose parent was deleted), counts
|
||||
as a dev workspace. Every job is covered: runs, previews, schedules, triggers, flow steps, AI
|
||||
agent tools and dependency jobs, and inline runs are refused on workers not serving dev
|
||||
workspaces. A job whose tag no serving worker listens to stays queued: it never falls back to
|
||||
the other worker groups. Dedicated workers keep their own tags.
|
||||
</p>
|
||||
|
||||
<!-- Worker groups serving dev workspaces -->
|
||||
<div class="flex flex-col gap-2">
|
||||
<span class="text-xs font-semibold text-emphasis">Worker groups serving dev workspaces</span>
|
||||
{#if flaggedGroups.length > 0}
|
||||
<div class="flex flex-wrap gap-2">
|
||||
{#each flaggedGroups as g (g.name)}
|
||||
<Button
|
||||
variant="default"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: Settings }}
|
||||
onClick={() => onConfigureGroup(g.name)}
|
||||
>
|
||||
{g.name}
|
||||
<span class={g.live > 0 ? 'text-secondary' : 'text-red-500'}>
|
||||
· {pluralize(g.live, 'live worker')}
|
||||
</span>
|
||||
</Button>
|
||||
{/each}
|
||||
</div>
|
||||
{:else}
|
||||
<span class="text-xs text-secondary">None yet.</span>
|
||||
{/if}
|
||||
{#if canEdit && unflaggedGroupNames.length > 0}
|
||||
<div class="flex gap-2 items-center">
|
||||
<Select
|
||||
size="sm"
|
||||
class="w-56"
|
||||
placeholder="Pick a worker group"
|
||||
items={unflaggedGroupNames.map((n) => ({ value: n }))}
|
||||
bind:value={groupToFlag}
|
||||
/>
|
||||
<Button
|
||||
variant="default"
|
||||
unifiedSize="sm"
|
||||
disabled={!groupToFlag}
|
||||
onClick={() => groupToFlag && onConfigureGroup(groupToFlag)}
|
||||
>
|
||||
Flag as serving dev workspaces
|
||||
</Button>
|
||||
<Tooltip>
|
||||
{#snippet text()}
|
||||
Opens the group's config, where "Serves dev workspaces and forks" sits under its tags.
|
||||
A dedicated group to start from: create a new group config, flag it, and start workers
|
||||
with its name as <code>WORKER_GROUP</code>.
|
||||
{/snippet}
|
||||
</Tooltip>
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
{#if enabled && liveFlaggedWorkers === 0}
|
||||
<Alert type="warning" size="xs" title="No worker serves dev workspaces">
|
||||
Jobs of dev workspaces and forks stay queued until a worker group flagged as serving dev
|
||||
workspaces has live workers{flaggedGroups.length === 0
|
||||
? ': flag a worker group as serving dev workspaces'
|
||||
: ''}.
|
||||
</Alert>
|
||||
{/if}
|
||||
|
||||
{#if unservedDevTags.length > 0}
|
||||
<Alert type="warning" size="xs" title="Dev jobs waiting with no worker">
|
||||
No live worker listens to
|
||||
{#each unservedDevTags as s, i (s.tag)}
|
||||
{i > 0 ? ', ' : ''}<code>{s.tag}</code> ({s.waiting} waiting)
|
||||
{/each}. These tags carry the <code>{effectiveSuffix}</code> suffix of dev workspace jobs: add
|
||||
the unsuffixed tag to a worker group flagged as serving dev workspaces, or flag a group that
|
||||
already lists it.
|
||||
</Alert>
|
||||
{/if}
|
||||
|
||||
{#if leakingGroups.length > 0}
|
||||
<Alert
|
||||
type="warning"
|
||||
size="xs"
|
||||
title="A worker group not serving dev workspaces takes dev jobs"
|
||||
>
|
||||
{#each leakingGroups as g, i (g.name)}
|
||||
{i > 0 ? '; ' : ''}<b>{g.name}</b> lists
|
||||
{#each g.tags as t, j (t)}{j > 0 ? ', ' : ''}<code>{t}</code>{/each}
|
||||
{/each}. Jobs of dev workspaces and forks run on these tags, so this group runs them next to
|
||||
root workspace jobs. Remove the tags, or flag the group as serving dev workspaces.
|
||||
</Alert>
|
||||
{/if}
|
||||
|
||||
<!-- Affected workspaces -->
|
||||
<div class="flex flex-col gap-1">
|
||||
<div class="w-fit">
|
||||
<Button
|
||||
variant="subtle"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: showWorkspaces ? ChevronDown : ChevronRight }}
|
||||
onClick={() => (showWorkspaces = !showWorkspaces)}
|
||||
>
|
||||
{#if workspaces === undefined}
|
||||
Workspaces routed to dev workers
|
||||
{:else}
|
||||
{pluralize(devWorkspaces.length, 'workspace')}
|
||||
{enabled ? 'routed' : 'that would be routed'} to dev workers
|
||||
{/if}
|
||||
</Button>
|
||||
</div>
|
||||
{#if showWorkspaces}
|
||||
{#if workspaces === undefined}
|
||||
<span class="text-xs text-secondary">Loading…</span>
|
||||
{:else if devWorkspaces.length === 0}
|
||||
<span class="text-xs text-secondary"
|
||||
>No dev workspace or fork yet. Only root workspaces exist.</span
|
||||
>
|
||||
{:else}
|
||||
<ul class="flex flex-col gap-1 text-xs max-h-60 overflow-auto">
|
||||
{#each devWorkspaces as w (w.id)}
|
||||
<li class="flex gap-2">
|
||||
<span class="font-mono text-primary">{w.id}</span>
|
||||
<span class="text-secondary">{workspaceKind(w)}</span>
|
||||
</li>
|
||||
{/each}
|
||||
</ul>
|
||||
{/if}
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
<!-- Advanced -->
|
||||
<div class="flex flex-col gap-2">
|
||||
<div class="w-fit">
|
||||
<Button
|
||||
variant="subtle"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: showAdvanced ? ChevronDown : ChevronRight }}
|
||||
onClick={() => (showAdvanced = !showAdvanced)}
|
||||
>
|
||||
Advanced
|
||||
</Button>
|
||||
</div>
|
||||
{#if showAdvanced}
|
||||
<div class="flex flex-col gap-1 max-w-md">
|
||||
<span class="text-xs font-semibold text-emphasis">Tag suffix</span>
|
||||
<span class="text-2xs text-secondary">
|
||||
Appended last, after every other tag setting: with workspace-specific default tags a
|
||||
fork's <code>python3-$workspace-fork</code> becomes
|
||||
<code>{withDevSuffix('python3-$workspace-fork', effectiveSuffix)}</code>. Changing it
|
||||
moves the worker groups serving dev workspaces to the new suffix right away, while jobs
|
||||
already queued keep the old one: change it when no dev job is waiting. A dev job whose
|
||||
tag would exceed 50 characters once suffixed is refused.
|
||||
</span>
|
||||
<div class="flex gap-2 items-start">
|
||||
<TextInput
|
||||
bind:value={suffixDraft}
|
||||
size="sm"
|
||||
error={suffixError}
|
||||
inputProps={{
|
||||
placeholder: DEFAULT_DEV_WORKSPACE_TAG_SUFFIX,
|
||||
disabled: !canEdit || !$enterpriseLicense
|
||||
}}
|
||||
/>
|
||||
<Button
|
||||
variant="accent"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: Save }}
|
||||
disabled={!canEdit ||
|
||||
!!suffixError ||
|
||||
suffixDraft.trim() === effectiveSuffix ||
|
||||
saving}
|
||||
onClick={() => saveSuffix(suffixDraft.trim())}
|
||||
>
|
||||
Save
|
||||
</Button>
|
||||
</div>
|
||||
{#if suffixError}
|
||||
<span class="text-2xs text-red-500">{suffixError}</span>
|
||||
{/if}
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
</div>
|
||||
</Section>
|
||||
@@ -28,7 +28,12 @@
|
||||
isRuleUnconditionallyActiveInRulesets,
|
||||
DEV_WORKSPACE_LOCK_RULE_NAME
|
||||
} from '$lib/workspaceProtectionRules.svelte'
|
||||
import { GitFork, ExternalLink, Check, Minus, Pen } from 'lucide-svelte'
|
||||
import { GitFork, ExternalLink, Check, Minus, Pen, Server } from 'lucide-svelte'
|
||||
import {
|
||||
devWorkspaceTagSuffix,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
import { resource } from 'runed'
|
||||
import type { Snippet } from 'svelte'
|
||||
|
||||
@@ -40,6 +45,7 @@
|
||||
deployTarget
|
||||
}: { deployTarget?: Snippet } = $props()
|
||||
|
||||
loadDevWorkspaceTagSuffix()
|
||||
let currentWs = $derived($userWorkspaces.find((w) => w.id === $workspaceStore))
|
||||
let isDev = $derived(currentWs?.is_dev_workspace ?? false)
|
||||
let currentLabel = $derived(devLabelKey(currentWs?.dev_workspace_label))
|
||||
@@ -455,6 +461,15 @@
|
||||
Pair this workspace with a dev workspace: the same code with a different environment
|
||||
(resource and variable values). Edits are made in the dev workspace and promoted here.
|
||||
</p>
|
||||
<p class="text-2xs text-secondary">
|
||||
Attaching an existing workspace copies nothing: it keeps its own resources and variables.
|
||||
Creating a new dev workspace copies this workspace's resources and variables, credentials
|
||||
included; point them at staging before developers use it.
|
||||
{#if $devWorkspaceTagSuffix}
|
||||
Either way, its jobs then run on dev workers (tags suffixed with
|
||||
<code>{$devWorkspaceTagSuffix}</code>).
|
||||
{/if}
|
||||
</p>
|
||||
<div class="flex flex-col gap-1">
|
||||
<span class="text-xs font-semibold text-emphasis">Attach an existing workspace as dev</span>
|
||||
<Select
|
||||
@@ -539,11 +554,26 @@
|
||||
{/if}
|
||||
{/snippet}
|
||||
|
||||
{#snippet devWorkersNotice()}
|
||||
{#if $devWorkspaceTagSuffix}
|
||||
<p class="text-2xs text-secondary max-w-2xl flex items-center gap-1.5">
|
||||
<Server size={12} />
|
||||
<span>
|
||||
Jobs in this workspace run on dev workers (tags suffixed with
|
||||
<code>{$devWorkspaceTagSuffix}</code>, e.g.
|
||||
<code>{withDevSuffix('python3', $devWorkspaceTagSuffix)}</code>), set by the instance's dev
|
||||
workspace isolation.
|
||||
</span>
|
||||
</p>
|
||||
{/if}
|
||||
{/snippet}
|
||||
|
||||
{#if isThrowawayFork}
|
||||
<p class="text-sm text-secondary max-w-2xl">
|
||||
Dev workspace pairing is not available on a throwaway fork. This workspace is a fork of
|
||||
<b>{parentId}</b>.
|
||||
</p>
|
||||
<div class="mt-2">{@render devWorkersNotice()}</div>
|
||||
{@render deployTarget?.()}
|
||||
{:else if isDev && parentId}
|
||||
<div class="flex flex-col gap-3 max-w-2xl">
|
||||
@@ -563,6 +593,7 @@
|
||||
<span class="font-mono">{currentLabel}</span> branch.
|
||||
</span>
|
||||
</div>
|
||||
{@render devWorkersNotice()}
|
||||
</div>
|
||||
{@render deployTarget?.()}
|
||||
<div class="flex flex-col gap-3 max-w-2xl mt-6">
|
||||
|
||||
@@ -56,6 +56,12 @@
|
||||
import DedicatedWorkersSelector from './DedicatedWorkersSelector.svelte'
|
||||
import { computeHashedTag } from './dedicated_worker'
|
||||
import { apiBaseUrl } from '$lib/apiBaseUrl.svelte'
|
||||
import {
|
||||
devWorkspaceListenSuffix,
|
||||
devWorkspaceTagSuffix,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
import ObjectStoreConfigSettings, {
|
||||
type ObjectStoreConfig
|
||||
} from './ObjectStoreConfigSettings.svelte'
|
||||
@@ -106,6 +112,7 @@
|
||||
native_mode?: boolean
|
||||
object_store_cache_config?: ObjectStoreConfig
|
||||
paused?: boolean
|
||||
serves_dev_workspaces?: boolean
|
||||
} = $state({})
|
||||
|
||||
function loadNConfig() {
|
||||
@@ -214,6 +221,7 @@
|
||||
native_mode?: boolean
|
||||
object_store_cache_config?: ObjectStoreConfig
|
||||
paused?: boolean
|
||||
serves_dev_workspaces?: boolean
|
||||
}
|
||||
activeWorkers: number
|
||||
customTags: string[] | undefined
|
||||
@@ -319,7 +327,8 @@
|
||||
'priority_tags',
|
||||
'additional_python_paths',
|
||||
'pip_local_dependencies',
|
||||
'object_store_cache_config'
|
||||
'object_store_cache_config',
|
||||
'serves_dev_workspaces'
|
||||
] as const
|
||||
function restartsWorkers(body: typeof nconfig): boolean {
|
||||
if (activeWorkers === 0) return false
|
||||
@@ -381,6 +390,13 @@
|
||||
!t.startsWith('flow-')
|
||||
)
|
||||
)
|
||||
loadDevWorkspaceTagSuffix()
|
||||
let devSuffix = $derived($devWorkspaceListenSuffix)
|
||||
let servesDev = $derived(
|
||||
config?.serves_dev_workspaces === true &&
|
||||
config?.dedicated_worker == undefined &&
|
||||
!config?.dedicated_workers?.length
|
||||
)
|
||||
let isAutoNativeMode = $derived(name === 'native')
|
||||
let isNativeModeEnabled = $derived(nconfig?.native_mode === true || isAutoNativeMode)
|
||||
$effect(() => {
|
||||
@@ -647,6 +663,63 @@
|
||||
{/if}
|
||||
</Label>
|
||||
|
||||
{#if nconfig !== undefined}
|
||||
{@const flagged = nconfig.serves_dev_workspaces === true}
|
||||
{@const draftTags = nconfig.worker_tags ?? []}
|
||||
{@const handSuffixed = draftTags.filter((t) => t.endsWith(devSuffix))}
|
||||
<div class="mt-8"></div>
|
||||
<Label label="Dev workspaces" eeOnly={!hasEnterpriseFeatures}>
|
||||
<Toggle
|
||||
size="sm"
|
||||
options={{
|
||||
right: 'Serves dev workspaces and forks',
|
||||
rightTooltip:
|
||||
'Runs jobs from dev workspaces and forks only. The group listens to each of its tags with the dev suffix, and no longer to the tags themselves.'
|
||||
}}
|
||||
checked={flagged}
|
||||
on:change={(ev) => {
|
||||
if (nconfig !== undefined) {
|
||||
nconfig.serves_dev_workspaces = ev.detail ? true : undefined
|
||||
}
|
||||
}}
|
||||
disabled={!canEditEEConfig}
|
||||
/>
|
||||
<p class="text-2xs text-secondary">Runs jobs from dev workspaces and forks only.</p>
|
||||
{#if flagged}
|
||||
<div class="flex flex-col gap-1 mt-1">
|
||||
<span class="text-2xs text-secondary">Tags this group listens to:</span>
|
||||
{#if draftTags.length > 0}
|
||||
<TagList
|
||||
tags={[...new Set(draftTags.map((t) => withDevSuffix(t, devSuffix)))]}
|
||||
class="flex-wrap"
|
||||
/>
|
||||
{:else}
|
||||
<span class="text-2xs text-secondary">
|
||||
Add tags above: each is listened to as <code
|
||||
>{withDevSuffix('tag', devSuffix)}</code
|
||||
>.
|
||||
</span>
|
||||
{/if}
|
||||
</div>
|
||||
{#if $devWorkspaceTagSuffix === null}
|
||||
<Alert size="xs" type="info" title="Dev workspace isolation is off">
|
||||
New jobs get the <code>{devSuffix}</code> suffix only while dev workspace isolation is
|
||||
on (Workers page). Until then this group only runs jobs already queued on suffixed tags,
|
||||
and never takes jobs of root workspaces.
|
||||
</Alert>
|
||||
{/if}
|
||||
{:else if handSuffixed.length > 0}
|
||||
<Alert size="xs" type="warning" title="This group would take dev workspace jobs">
|
||||
{handSuffixed.join(', ')}
|
||||
{handSuffixed.length === 1 ? 'is a tag' : 'are tags'} of dev workspace and fork jobs. Listing
|
||||
{handSuffixed.length === 1 ? 'it' : 'them'} here runs those jobs next to root workspace
|
||||
jobs. Remove {handSuffixed.length === 1 ? 'it' : 'them'}, or flag this group as
|
||||
serving dev workspaces instead.
|
||||
</Alert>
|
||||
{/if}
|
||||
</Label>
|
||||
{/if}
|
||||
|
||||
{#if nconfig?.worker_tags !== undefined && nconfig?.worker_tags.length > 0}
|
||||
<div class="mt-8"></div>
|
||||
<Label label="High-priority tags" eeOnly={!hasEnterpriseFeatures}>
|
||||
@@ -1364,6 +1437,17 @@
|
||||
{#if isNativeMode}
|
||||
<Badge color="blue" small>Native</Badge>
|
||||
{/if}
|
||||
{#if servesDev}
|
||||
<Badge color="violet" small>
|
||||
Dev
|
||||
<Tooltip>
|
||||
{#snippet text()}
|
||||
Serves dev workspaces and forks only: listens to its tags suffixed with
|
||||
<code>{devSuffix}</code>, never to the tags of root workspace jobs.
|
||||
{/snippet}
|
||||
</Tooltip>
|
||||
</Badge>
|
||||
{/if}
|
||||
{#if config?.paused === true}
|
||||
<Badge color="yellow" small>
|
||||
Paused
|
||||
@@ -1521,7 +1605,13 @@
|
||||
{#if config?.worker_tags && config.worker_tags.length > 0}
|
||||
<div class="flex flex-row items-start gap-2 w-full">
|
||||
<div class="text-secondary text-xs mt-1">Tags:</div>
|
||||
<TagList tags={config.worker_tags} maxVisible={25} class="flex-wrap" />
|
||||
<TagList
|
||||
tags={servesDev
|
||||
? [...new Set(config.worker_tags.map((t) => withDevSuffix(t, devSuffix)))]
|
||||
: config.worker_tags}
|
||||
maxVisible={25}
|
||||
class="flex-wrap"
|
||||
/>
|
||||
</div>
|
||||
{:else if config?.dedicated_workers && config.dedicated_workers.length > 0}
|
||||
<div class="flex flex-row items-start gap-2 w-full min-w-0">
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<script lang="ts">
|
||||
import { workerTags, workspaceStore } from '$lib/stores'
|
||||
import { userWorkspaces, workerTags, workspaceStore } from '$lib/stores'
|
||||
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
|
||||
import { WorkerService } from '$lib/gen'
|
||||
|
||||
@@ -10,6 +10,12 @@
|
||||
import { RotateCw } from 'lucide-svelte'
|
||||
import { sendUserToast } from '$lib/toast'
|
||||
import Popover from './Popover.svelte'
|
||||
import {
|
||||
devWorkspaceTagSuffix,
|
||||
isDevOrForkWorkspaceId,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
|
||||
let {
|
||||
tag = $bindable(),
|
||||
@@ -113,10 +119,16 @@
|
||||
return
|
||||
}
|
||||
if (open) {
|
||||
tagsToWorkerExists = await WorkerService.existsWorkersWithTags({
|
||||
tags: items.filter((t) => !dynamicTagRegex.test(t)).join(','),
|
||||
// Probed as jobs of this workspace land: suffixed in a dev workspace.
|
||||
const probed = items.filter((t) => !dynamicTagRegex.test(t))
|
||||
const asRun = (t: string) => (devSuffix ? withDevSuffix(t, devSuffix) : t)
|
||||
const exists = await WorkerService.existsWorkersWithTags({
|
||||
tags: probed.map(asRun).join(','),
|
||||
workspace: effectiveWorkspace
|
||||
})
|
||||
tagsToWorkerExists = Object.fromEntries(
|
||||
probed.filter((t) => asRun(t) in exists).map((t) => [t, exists[asRun(t)]])
|
||||
)
|
||||
lastCheck = Date.now()
|
||||
if (visible) {
|
||||
timeout = setTimeout(loadTagsToWorkerExists, 5000)
|
||||
@@ -145,6 +157,13 @@
|
||||
})
|
||||
|
||||
let open = $state(false)
|
||||
|
||||
loadDevWorkspaceTagSuffix()
|
||||
let devSuffix = $derived(
|
||||
$devWorkspaceTagSuffix && isDevOrForkWorkspaceId(effectiveWorkspace, $userWorkspaces)
|
||||
? $devWorkspaceTagSuffix
|
||||
: undefined
|
||||
)
|
||||
</script>
|
||||
|
||||
{#snippet startSnippet({ item })}
|
||||
@@ -186,13 +205,29 @@
|
||||
{startSnippet}
|
||||
bottomSnippet={refreshAll}
|
||||
/>
|
||||
{#if devSuffix && tag}
|
||||
<Popover>
|
||||
{#snippet text()}
|
||||
This workspace is a dev workspace or a fork, so its jobs run on worker groups serving dev
|
||||
workspaces, with every tag suffixed with <code>{devSuffix}</code>.
|
||||
{/snippet}
|
||||
<span class="text-2xs text-hint whitespace-nowrap">
|
||||
runs as <code>{withDevSuffix(tag, devSuffix)}</code> here
|
||||
</span>
|
||||
</Popover>
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
{#snippet refreshAll()}
|
||||
<div class="flex items-center justify-between gap-2 border-t border-border-light">
|
||||
<span class="max-w-64 px-4 py-1 text-2xs text-secondary">
|
||||
<code>$workspace</code>, <code>$args[…]</code> and <code>$flow_expr[…]</code> are filled in when
|
||||
the job runs, and must land on an allowed tag
|
||||
<code>$workspace</code>, <code>$args[…]</code> and <code>$flow_expr[…]</code> are filled in
|
||||
when the job runs, and must land on an allowed tag
|
||||
{#if devSuffix}
|
||||
<br />In this dev workspace every tag runs suffixed with <code>{devSuffix}</code>, the
|
||||
default one included: <code>gpu</code> runs as
|
||||
<code>{withDevSuffix('gpu', devSuffix)}</code>
|
||||
{/if}
|
||||
</span>
|
||||
<Button
|
||||
iconOnly
|
||||
|
||||
@@ -4,6 +4,13 @@
|
||||
import { resource } from 'runed'
|
||||
import { hasWorkerForTag } from './missingWorker'
|
||||
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
|
||||
import { userWorkspaces } from '$lib/stores'
|
||||
import {
|
||||
devWorkspaceTagSuffix,
|
||||
isDevOrForkWorkspaceId,
|
||||
loadDevWorkspaceTagSuffix,
|
||||
withDevSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
|
||||
const operatingWorkspace = useOperatingWorkspace()
|
||||
|
||||
@@ -24,16 +31,33 @@
|
||||
// `tag`, so probing the bare tag would warn about a tag nothing uses. The
|
||||
// in-flight check in `pollJobResult` reads the job's real tag and still covers
|
||||
// those instances.
|
||||
// A dev workspace's jobs run on the suffixed tag, which only groups serving dev workspaces pull.
|
||||
let devSuffix = $derived(
|
||||
$devWorkspaceTagSuffix && isDevOrForkWorkspaceId(ws, $userWorkspaces)
|
||||
? $devWorkspaceTagSuffix
|
||||
: undefined
|
||||
)
|
||||
let runTag = $derived(devSuffix ? withDevSuffix(tag, devSuffix) : tag)
|
||||
const served = resource(
|
||||
() => [ws, tag] as const,
|
||||
async ([ws, tag]) => {
|
||||
() => [ws, runTag] as const,
|
||||
async ([ws, runTag]) => {
|
||||
await loadDevWorkspaceTagSuffix()
|
||||
if (!ws || (await WorkerService.isDefaultTagsPerWorkspace())) return true
|
||||
return await hasWorkerForTag(ws, tag)
|
||||
return await hasWorkerForTag(ws, runTag)
|
||||
}
|
||||
)
|
||||
</script>
|
||||
|
||||
{#if served.current === false}
|
||||
{#if served.current === false && devSuffix}
|
||||
<div class={className}>
|
||||
<Alert type="warning" title="No worker is running for the "{runTag}" tag" size="xs">
|
||||
{subject} run as Windmill jobs tagged <b>{tag}</b>, which this workspace, a dev workspace or a
|
||||
fork, runs as <b>{runTag}</b>. No worker is currently listening to it, so they stay queued
|
||||
until one is. An admin can serve it by flagging a worker group that lists <b>{tag}</b> as
|
||||
serving dev workspaces on the <a href="/workers">workers page</a>.
|
||||
</Alert>
|
||||
</div>
|
||||
{:else if served.current === false}
|
||||
<div class={className}>
|
||||
<!-- Deliberately not phrased as "this tag is not configured": a correctly
|
||||
tagged group sitting at zero replicas is indistinguishable from one in
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { JobService, WorkerService } from '$lib/gen'
|
||||
import { devTagNoWorkerHint, loadDevWorkspaceTagSuffix } from '$lib/devWorkspaceTags'
|
||||
|
||||
/**
|
||||
* A queued job whose tag no running worker serves is never picked up: without
|
||||
@@ -15,9 +16,10 @@ export class NoWorkerForTagError extends Error {
|
||||
constructor(tag: string) {
|
||||
super(
|
||||
`No worker has been listening to the tag "${tag}" while this job waited, so it was never ` +
|
||||
`picked up. It stays queued and will run once a worker with that tag comes online. ` +
|
||||
`Add "${tag}" to the worker tags of one of your worker groups (Workers page), or run a ` +
|
||||
`worker that serves it.`
|
||||
`picked up. It stays queued and will run once a worker with that tag comes online.` +
|
||||
(devTagNoWorkerHint(tag) ||
|
||||
` Add "${tag}" to the worker tags of one of your worker groups (Workers page), or run ` +
|
||||
`a worker that serves it.`)
|
||||
)
|
||||
this.name = 'NoWorkerForTagError'
|
||||
this.tag = tag
|
||||
@@ -28,8 +30,10 @@ export class NoWorkerForTagError extends Error {
|
||||
export function queuedWithoutWorkerMessage(tag: string): string {
|
||||
return (
|
||||
`No worker is listening to the tag "${tag}", so this operation is queued and will only run ` +
|
||||
`once one is. Add "${tag}" to the worker tags of one of your worker groups (Workers page), ` +
|
||||
`or run a worker that serves it.`
|
||||
`once one is.` +
|
||||
(devTagNoWorkerHint(tag) ||
|
||||
` Add "${tag}" to the worker tags of one of your worker groups (Workers page), or run a ` +
|
||||
`worker that serves it.`)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -68,7 +72,9 @@ export async function missingWorkerTagOfQueuedJob(
|
||||
try {
|
||||
const job = await JobService.getJob({ workspace, id: jobId, noCode: true, noLogs: true })
|
||||
if (job.type !== 'QueuedJob' || job.running || !job.tag) return undefined
|
||||
return (await hasWorkerForTag(workspace, job.tag)) ? undefined : job.tag
|
||||
if (await hasWorkerForTag(workspace, job.tag)) return undefined
|
||||
await loadDevWorkspaceTagSuffix()
|
||||
return job.tag
|
||||
} catch (err) {
|
||||
console.warn('Could not determine whether a worker serves the job tag', err)
|
||||
return undefined
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
import Popover from '../Popover.svelte'
|
||||
import { onDestroy, untrack } from 'svelte'
|
||||
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
|
||||
import { devWorkspaceTagSuffix, loadDevWorkspaceTagSuffix } from '$lib/devWorkspaceTags'
|
||||
|
||||
const operatingWorkspace = useOperatingWorkspace()
|
||||
interface Props {
|
||||
@@ -28,6 +29,13 @@
|
||||
return tag
|
||||
})
|
||||
|
||||
loadDevWorkspaceTagSuffix()
|
||||
let devSuffix = $derived(
|
||||
$devWorkspaceTagSuffix && customTag?.endsWith($devWorkspaceTagSuffix)
|
||||
? $devWorkspaceTagSuffix
|
||||
: undefined
|
||||
)
|
||||
|
||||
async function lookForTag(): Promise<void> {
|
||||
try {
|
||||
if (!customTag) return
|
||||
@@ -50,7 +58,10 @@
|
||||
|
||||
$effect(() => {
|
||||
customTag
|
||||
untrack(() => timeout && setTimeout(() => lookForTag(), 2500))
|
||||
untrack(() => {
|
||||
clearTimeout(timeout)
|
||||
timeout = setTimeout(() => lookForTag(), 2500)
|
||||
})
|
||||
})
|
||||
|
||||
onDestroy(() => {
|
||||
@@ -66,6 +77,10 @@
|
||||
<AlertTriangle size={16} class="text-yellow-500" />
|
||||
{#snippet text()}
|
||||
No worker with {tagLabel ?? 'tag'} <b>{customTag}</b> is currently running.
|
||||
{#if devSuffix}
|
||||
<br />It carries the <code>{devSuffix}</code> suffix of dev workspace and fork jobs: only worker
|
||||
groups flagged as serving dev workspaces listen to it. An admin can flag one on the Workers page.
|
||||
{/if}
|
||||
{/snippet}
|
||||
</Popover>
|
||||
{/if}
|
||||
|
||||
@@ -8,8 +8,15 @@
|
||||
import { ListFilter, Tag } from 'lucide-svelte'
|
||||
import type { Job } from '$lib/gen'
|
||||
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
|
||||
import { userWorkspaces } from '$lib/stores'
|
||||
import {
|
||||
devWorkspaceTagSuffix,
|
||||
isDevOrForkWorkspaceId,
|
||||
loadDevWorkspaceTagSuffix
|
||||
} from '$lib/devWorkspaceTags'
|
||||
|
||||
const operatingWorkspace = useOperatingWorkspace()
|
||||
loadDevWorkspaceTagSuffix()
|
||||
|
||||
interface Props {
|
||||
job: Job
|
||||
@@ -67,7 +74,17 @@
|
||||
{#if job.tag != undefined}
|
||||
<!-- for related places search: ADD_NEW_LANG -->
|
||||
<div>
|
||||
<Badge color="indigo" {large}>Tag: {job.tag}</Badge>
|
||||
<Badge color="indigo" {large}>
|
||||
Tag: {job.tag}
|
||||
{#if $devWorkspaceTagSuffix && job.tag.endsWith($devWorkspaceTagSuffix) && isDevOrForkWorkspaceId(job.workspace_id, $userWorkspaces)}
|
||||
<Tooltip>
|
||||
{#snippet text()}
|
||||
Suffixed with <code>{$devWorkspaceTagSuffix}</code> because this workspace is a dev workspace
|
||||
or a fork: its jobs run only on worker groups serving dev workspaces.
|
||||
{/snippet}
|
||||
</Tooltip>
|
||||
{/if}
|
||||
</Badge>
|
||||
</div>
|
||||
{/if}
|
||||
{#if !job.visible_to_owner}
|
||||
@@ -107,9 +124,7 @@
|
||||
<ListFilter class="inline-block" size={10} />
|
||||
</Button>
|
||||
{:else}
|
||||
<a
|
||||
href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}
|
||||
>
|
||||
<a href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}>
|
||||
{concurrencyKey}
|
||||
</a>
|
||||
{/if}
|
||||
@@ -119,9 +134,7 @@
|
||||
<Badge {large}>Concurrency: {truncateRev(concurrencyKey, 20)}</Badge>
|
||||
</button>
|
||||
{:else}
|
||||
<a
|
||||
href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}
|
||||
>
|
||||
<a href={`/runs/?job_kinds=all&graph=ConcurrencyChart&concurrency_key=${concurrencyKey}`}>
|
||||
<Badge {large}>Concurrency: {truncateRev(concurrencyKey, 20)}</Badge>
|
||||
</a>
|
||||
{/if}
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
type CompletedJob
|
||||
} from '$lib/gen'
|
||||
import { validateUsername } from '$lib/utils'
|
||||
import { devWorkspaceTagSuffix, loadDevWorkspaceTagSuffix } from '$lib/devWorkspaceTags'
|
||||
import { logoutWithRedirect } from '$lib/logoutKit'
|
||||
import { page } from '$app/state'
|
||||
import {
|
||||
@@ -162,6 +163,12 @@
|
||||
// read as a root (it would offer invalid dev creation while the workspace list is still loading).
|
||||
// `workspaceIsFork` (prefix OR parent) also excludes an orphaned `wm-fork-` workspace, whose parent
|
||||
// FK was set null — it has no parent but is still a fork, so it can't host a dev workspace.
|
||||
let baseIsRoot = $derived(
|
||||
!!baseWorkspaceEntry && !workspaceIsFork(baseWorkspaceId, forkableWorkspaces)
|
||||
)
|
||||
$effect(() => {
|
||||
isFork && loadDevWorkspaceTagSuffix()
|
||||
})
|
||||
let baseCanHostDev = $derived(
|
||||
!!baseWorkspaceEntry &&
|
||||
(!workspaceIsFork(baseWorkspaceId, forkableWorkspaces) ||
|
||||
@@ -963,6 +970,34 @@
|
||||
there.
|
||||
</Alert>
|
||||
{/if}
|
||||
{#if baseWorkspaceId && createAsDevWorkspace}
|
||||
<Alert
|
||||
type="info"
|
||||
size="xs"
|
||||
title="Resources and variables are copied from {baseWorkspaceId}"
|
||||
>
|
||||
The dev workspace starts with copies of {baseWorkspaceId}'s resources and variables,
|
||||
credentials included. Point them at staging before developers use it.
|
||||
{#if !forkingLocked && !lockProdForking}
|
||||
Turn on <b>Prevent forking</b> above so developers fork this dev workspace, which will
|
||||
hold staging credentials, rather than {baseWorkspaceId}.
|
||||
{/if}
|
||||
{#if $devWorkspaceTagSuffix}
|
||||
Its jobs run on dev workers (tags suffixed with <code>{$devWorkspaceTagSuffix}</code
|
||||
>).
|
||||
{/if}
|
||||
</Alert>
|
||||
{:else if baseWorkspaceId && $devWorkspaceTagSuffix}
|
||||
<Alert type="info" size="xs" title="Runs on dev workers">
|
||||
Jobs of this fork run on the worker groups serving dev workspaces (tags suffixed with
|
||||
<code>{$devWorkspaceTagSuffix}</code>).
|
||||
{#if baseIsRoot}
|
||||
It copies {baseWorkspaceId}'s resources and variables, so its scripts use
|
||||
{baseWorkspaceId}'s credentials from those workers and reach what their network
|
||||
allows. Forking a dev workspace instead gives a fork its staging credentials.
|
||||
{/if}
|
||||
</Alert>
|
||||
{/if}
|
||||
{/if}
|
||||
{#if isFork}
|
||||
<ForkDatatableSection
|
||||
@@ -1101,11 +1136,8 @@
|
||||
</div>
|
||||
<div class="flex flex-wrap flex-row {inModal ? 'justify-end' : 'justify-between'} gap-4 pt-4">
|
||||
{#if !inModal}
|
||||
<Button
|
||||
disabled={forkCreationLoading}
|
||||
variant="default"
|
||||
size="sm"
|
||||
href="/user/workspaces">← Back to workspaces</Button
|
||||
<Button disabled={forkCreationLoading} variant="default" size="sm" href="/user/workspaces"
|
||||
>← Back to workspaces</Button
|
||||
>
|
||||
{/if}
|
||||
{#if !forkCreationLoading}
|
||||
|
||||
@@ -48,6 +48,8 @@ export const DEFAULT_TAGS_WORKSPACES_SETTING = 'default_tags_workspaces'
|
||||
export const FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING = 'fork_workspace_tag_append_fork_suffix'
|
||||
export const PREVIEW_TAGS_OVERRIDE_SETTING = 'preview_tags_override'
|
||||
export const DEPENDENCY_JOB_TAG_SETTING = 'dependency_job_tag'
|
||||
export const DEV_WORKSPACE_ISOLATION_SETTING = 'dev_workspace_isolation'
|
||||
export const DEV_WORKSPACE_TAG_SUFFIX_SETTING = 'dev_workspace_tag_suffix'
|
||||
|
||||
export const WORKSPACE_FAIRNESS_ENABLED_SETTING = 'workspace_fairness_enabled'
|
||||
export const WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING = 'workspace_fairness_max_percent'
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import { derived, get, writable } from 'svelte/store'
|
||||
import { WorkerService } from '$lib/gen'
|
||||
|
||||
/** Mirrors `DEFAULT_DEV_WORKSPACE_TAG_SUFFIX` in `windmill-common/src/worker.rs`. */
|
||||
export const DEFAULT_DEV_WORKSPACE_TAG_SUFFIX = '-dev'
|
||||
const WM_FORK_PREFIX = 'wm-fork-'
|
||||
|
||||
/** Mirrors `is_valid_dev_workspace_tag_suffix`: the suffix ends up in SQL literals and tag lists. */
|
||||
export const DEV_WORKSPACE_TAG_SUFFIX_PATTERN = /^[A-Za-z0-9_.-]{1,32}$/
|
||||
|
||||
/** `undefined` until loaded. */
|
||||
export const devWorkspaceIsolation = writable<{ enabled: boolean; suffix: string } | undefined>(
|
||||
undefined
|
||||
)
|
||||
/** The suffix new dev workspace and fork jobs get: `null` while isolation is off. */
|
||||
export const devWorkspaceTagSuffix = derived(devWorkspaceIsolation, (i) =>
|
||||
i === undefined ? undefined : i.enabled ? i.suffix : null
|
||||
)
|
||||
/** The suffix worker groups serving dev workspaces listen with, isolation on or off. */
|
||||
export const devWorkspaceListenSuffix = derived(
|
||||
devWorkspaceIsolation,
|
||||
(i) => i?.suffix ?? DEFAULT_DEV_WORKSPACE_TAG_SUFFIX
|
||||
)
|
||||
|
||||
let inflight: Promise<void> | undefined
|
||||
export function loadDevWorkspaceTagSuffix(force = false): Promise<void> {
|
||||
if (!force && get(devWorkspaceIsolation) !== undefined) return Promise.resolve()
|
||||
if (!force && inflight) return inflight
|
||||
// Started inside the chain so any failure lands in the catch: callers on a job's error path
|
||||
// await this and must not fail because the setting could not be read.
|
||||
inflight = Promise.resolve()
|
||||
.then(() => WorkerService.getDevWorkspaceIsolation())
|
||||
.then((i) => devWorkspaceIsolation.set(i))
|
||||
.catch(() =>
|
||||
devWorkspaceIsolation.set({ enabled: false, suffix: DEFAULT_DEV_WORKSPACE_TAG_SUFFIX })
|
||||
)
|
||||
.finally(() => (inflight = undefined))
|
||||
return inflight
|
||||
}
|
||||
|
||||
/** Mirrors `with_dev_workspace_suffix`: idempotent, as an inherited tag already carries it. */
|
||||
export function withDevSuffix(tag: string, suffix: string): string {
|
||||
return tag.endsWith(suffix) ? tag : `${tag}${suffix}`
|
||||
}
|
||||
|
||||
/** The suffix `tag` carries as a dev workspace job's tag, if isolation is on and it does. */
|
||||
export function devSuffixOf(tag: string | undefined): string | undefined {
|
||||
const suffix = get(devWorkspaceTagSuffix)
|
||||
return suffix && tag?.endsWith(suffix) ? suffix : undefined
|
||||
}
|
||||
|
||||
/** What to do about a dev workspace tag no worker serves, empty for any other tag. */
|
||||
export function devTagNoWorkerHint(tag: string): string {
|
||||
const suffix = devSuffixOf(tag)
|
||||
if (!suffix) return ''
|
||||
return (
|
||||
` "${tag}" ends with "${suffix}", the suffix of jobs of dev workspaces and forks: only worker ` +
|
||||
`groups flagged as serving dev workspaces listen to it. Flag a worker group as serving dev ` +
|
||||
`workspaces (Workers page) whose tags include "${tag.slice(0, -suffix.length)}".`
|
||||
)
|
||||
}
|
||||
|
||||
/** Whether a workspace's jobs run on dev workers: it has a parent, or is (or was) a fork. */
|
||||
export function isDevOrForkWorkspace(w: {
|
||||
id: string
|
||||
parent_workspace_id?: string | null
|
||||
}): boolean {
|
||||
return !!w.parent_workspace_id || w.id.startsWith(WM_FORK_PREFIX)
|
||||
}
|
||||
|
||||
/** The same check for a workspace id, looked up in `workspaces` (pass the store's value so a
|
||||
* reactive caller recomputes once the list loads). */
|
||||
export function isDevOrForkWorkspaceId(
|
||||
workspaceId: string | undefined,
|
||||
workspaces: { id: string; parent_workspace_id?: string | null }[]
|
||||
): boolean {
|
||||
if (!workspaceId) return false
|
||||
return isDevOrForkWorkspace(workspaces.find((w) => w.id === workspaceId) ?? { id: workspaceId })
|
||||
}
|
||||
@@ -16,6 +16,8 @@
|
||||
import Head from '$lib/components/table/Head.svelte'
|
||||
import Tooltip from '$lib/components/Tooltip.svelte'
|
||||
import WorkspaceGroup from '$lib/components/WorkerGroup.svelte'
|
||||
import DevWorkspaceIsolation from '$lib/components/DevWorkspaceIsolation.svelte'
|
||||
import { devWorkspaceListenSuffix, withDevSuffix } from '$lib/devWorkspaceTags'
|
||||
import {
|
||||
WorkerService,
|
||||
type WorkerPing,
|
||||
@@ -904,6 +906,23 @@
|
||||
</Alert>
|
||||
{/if}
|
||||
|
||||
{#if $superadmin || $devopsRole}
|
||||
<DevWorkspaceIsolation
|
||||
{workerGroups}
|
||||
liveWorkers={Object.fromEntries(
|
||||
groupedWorkers.map(([name, instances]) => [
|
||||
name,
|
||||
instances.flatMap(([_, pings]) => pings.filter((p) => (p.last_ping ?? 0) < 15)).length
|
||||
])
|
||||
)}
|
||||
canEdit={!!$superadmin}
|
||||
onConfigureGroup={(name) => {
|
||||
selectedTab = name
|
||||
shouldAutoOpenDrawer = name
|
||||
}}
|
||||
/>
|
||||
{/if}
|
||||
|
||||
{#if workers != undefined}
|
||||
{#if groupedWorkers.length == 0}
|
||||
<p>No workers seem to be available</p>
|
||||
@@ -1046,7 +1065,13 @@
|
||||
{@const isWorkerAlive = isWorkerMaybeAlive(last_ping)}
|
||||
{@const tagMismatchInfo = getTagMismatchInfo(
|
||||
custom_tags,
|
||||
config?.worker_tags
|
||||
config?.serves_dev_workspaces === true &&
|
||||
config?.dedicated_worker == undefined &&
|
||||
!config?.dedicated_workers?.length
|
||||
? config?.worker_tags?.map((t: string) =>
|
||||
withDevSuffix(t, $devWorkspaceListenSuffix)
|
||||
)
|
||||
: config?.worker_tags
|
||||
)}
|
||||
<tr class={groupIdx % 2 == 1 ? 'bg-surface-secondary/50' : ''}>
|
||||
<Cell class="text-primary" first>
|
||||
|
||||
Reference in New Issue
Block a user