mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-03 08:02:19 +00:00
feat: add an instance setting routing all dependency jobs to one tag (#11486)
* feat: add an instance setting routing all dependency jobs to one tag Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: keep bunnative locks on bun and explain the default dependency routing Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
14801fdd68
commit
3d1d249556
+39
-39
@@ -44,28 +44,27 @@ 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,
|
||||
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,
|
||||
DEPENDENCY_JOB_TAG_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,
|
||||
HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING,
|
||||
INSTANCE_EVENTS_WEBHOOK_SETTING, INSTANCE_PYTHON_VERSION_SETTING,
|
||||
JOB_DEFAULT_TIMEOUT_SECS_SETTING, JOB_ISOLATION_SETTING, JWT_SECRET_SETTING,
|
||||
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING, MAVEN_SETTINGS_XML_SETTING,
|
||||
MCP_DISABLE_TOKEN_QUERY_PARAM_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING,
|
||||
NO_DEFAULT_MAVEN_SETTING,
|
||||
NPM_CONFIG_REGISTRY_SETTING, NSJAIL_TMPFS_SIZE_MB_SETTING, NSJAIL_TMP_BACKING_SETTING,
|
||||
NUGET_CONFIG_SETTING, OAUTH_SETTING, OTEL_SETTING, OTEL_TRACES_RETENTION_SECS_SETTING,
|
||||
OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING,
|
||||
POWERSHELL_REPO_URL_SETTING, PREVIEW_TAGS_OVERRIDE_SETTING, REQUEST_SIZE_LIMIT_SETTING,
|
||||
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RESTART_COORDINATION_SETTING,
|
||||
RETENTION_PERIOD_SECS_OVERRIDES_SETTING, RETENTION_PERIOD_SECS_SETTING, RUBY_REPOS_SETTING,
|
||||
SAML_METADATA_SETTING, SANDBOX_IMAGE_CACHE_MAX_MB_SETTING,
|
||||
SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING, SANDBOX_IMAGE_MAX_SIZE_MB_SETTING,
|
||||
SANDBOX_IMAGE_PULL_POLICY_SETTING, SANDBOX_REGISTRY_AUTH_SETTING, SCIM_TOKEN_SETTING,
|
||||
SERVICE_LOG_RETENTION_SECS_SETTING, SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING,
|
||||
TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING,
|
||||
UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
|
||||
NO_DEFAULT_MAVEN_SETTING, NPM_CONFIG_REGISTRY_SETTING, NSJAIL_TMPFS_SIZE_MB_SETTING,
|
||||
NSJAIL_TMP_BACKING_SETTING, NUGET_CONFIG_SETTING, OAUTH_SETTING, OTEL_SETTING,
|
||||
OTEL_TRACES_RETENTION_SECS_SETTING, OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING,
|
||||
POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING, PREVIEW_TAGS_OVERRIDE_SETTING,
|
||||
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
|
||||
RESTART_COORDINATION_SETTING, RETENTION_PERIOD_SECS_OVERRIDES_SETTING,
|
||||
RETENTION_PERIOD_SECS_SETTING, RUBY_REPOS_SETTING, SAML_METADATA_SETTING,
|
||||
SANDBOX_IMAGE_CACHE_MAX_MB_SETTING, SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING,
|
||||
SANDBOX_IMAGE_MAX_SIZE_MB_SETTING, SANDBOX_IMAGE_PULL_POLICY_SETTING,
|
||||
SANDBOX_REGISTRY_AUTH_SETTING, SCIM_TOKEN_SETTING, SERVICE_LOG_RETENTION_SECS_SETTING,
|
||||
SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
|
||||
UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
|
||||
WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
|
||||
WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
|
||||
WORKSPACE_MAX_QUEUED_JOBS_SETTING, WORKSPACE_REGISTRIES_SETTING,
|
||||
@@ -126,24 +125,23 @@ use windmill_worker::{
|
||||
};
|
||||
|
||||
use crate::monitor::{
|
||||
initial_load, load_concurrency_key_max_queued, 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, load_workspace_fairness_enabled,
|
||||
load_workspace_fairness_max_percent, load_workspace_fairness_min_total,
|
||||
load_workspace_max_queued_jobs, monitor_db, reload_app_workspaced_route_setting,
|
||||
reload_audit_log_retention_days_setting, reload_base_url_setting,
|
||||
reload_bun_install_min_release_age_setting, reload_bunfig_install_scopes_setting,
|
||||
reload_critical_alert_mute_ui_setting, reload_critical_alert_mute_zombie_job_restart_setting,
|
||||
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_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,
|
||||
load_workspace_fairness_enabled, load_workspace_fairness_max_percent,
|
||||
load_workspace_fairness_min_total, load_workspace_max_queued_jobs, monitor_db,
|
||||
reload_app_workspaced_route_setting, reload_audit_log_retention_days_setting,
|
||||
reload_base_url_setting, reload_bun_install_min_release_age_setting,
|
||||
reload_bunfig_install_scopes_setting, reload_critical_alert_mute_ui_setting,
|
||||
reload_critical_alert_mute_zombie_job_restart_setting,
|
||||
reload_critical_alerts_on_token_expiry_setting, reload_critical_error_channels_setting,
|
||||
reload_extra_pip_index_url_setting, reload_http_route_default_allowed_origins_setting,
|
||||
reload_http_route_workspaced_route_setting,
|
||||
reload_hub_api_secret_setting, reload_hub_base_url_setting,
|
||||
reload_instance_events_webhook_setting, reload_job_default_timeout_setting,
|
||||
reload_job_isolation_setting, reload_jwt_secret_setting, reload_license_key,
|
||||
reload_npm_config_registry_setting, reload_nsjail_tmp_backing_setting,
|
||||
reload_http_route_workspaced_route_setting, reload_hub_api_secret_setting,
|
||||
reload_hub_base_url_setting, reload_instance_events_webhook_setting,
|
||||
reload_job_default_timeout_setting, reload_job_isolation_setting, reload_jwt_secret_setting,
|
||||
reload_license_key, reload_npm_config_registry_setting, reload_nsjail_tmp_backing_setting,
|
||||
reload_nsjail_tmpfs_size_setting, reload_otel_traces_retention_secs_setting,
|
||||
reload_otel_tracing_proxy_setting, reload_pip_index_url_setting,
|
||||
reload_retention_period_setting, reload_sandbox_image_cache_max_setting,
|
||||
@@ -423,7 +421,6 @@ struct HubResourceTypeRaw {
|
||||
pub display_name: Option<Option<String>>,
|
||||
}
|
||||
|
||||
|
||||
/// Processed resource type with parsed schema
|
||||
#[derive(serde::Deserialize, serde::Serialize, Clone)]
|
||||
pub struct HubResourceType {
|
||||
@@ -1889,12 +1886,10 @@ async fn process_notify_event(
|
||||
);
|
||||
windmill_common::workspaces::PUBLIC_APP_RATE_LIMIT_CACHE.remove(payload);
|
||||
}
|
||||
windmill_common::SCRIPT_VERSION_DELETED_CHANNEL => {
|
||||
match serde_json::from_str(payload) {
|
||||
Ok(deleted) => windmill_api_scripts::scripts::evict_deleted_script_versions(deleted),
|
||||
Err(e) => tracing::error!("Invalid script version deletion payload {payload}: {e}"),
|
||||
}
|
||||
}
|
||||
windmill_common::SCRIPT_VERSION_DELETED_CHANNEL => match serde_json::from_str(payload) {
|
||||
Ok(deleted) => windmill_api_scripts::scripts::evict_deleted_script_versions(deleted),
|
||||
Err(e) => tracing::error!("Invalid script version deletion payload {payload}: {e}"),
|
||||
},
|
||||
"notify_runnable_version_change" => {
|
||||
tracing::info!("Runnable version change detected: {}", payload);
|
||||
match payload.split(':').collect::<Vec<&str>>().as_slice() {
|
||||
@@ -2055,6 +2050,11 @@ async fn process_notify_event(
|
||||
tracing::error!("Error loading preview tags override: {e:#}");
|
||||
}
|
||||
}
|
||||
DEPENDENCY_JOB_TAG_SETTING => {
|
||||
if let Err(e) = load_dependency_job_tag(db).await {
|
||||
tracing::error!("Error loading dependency job tag: {e:#}");
|
||||
}
|
||||
}
|
||||
WORKSPACE_FAIRNESS_ENABLED_SETTING => {
|
||||
if let Err(e) = load_workspace_fairness_enabled(db).await {
|
||||
tracing::error!("Error loading workspace fairness enabled: {e:#}");
|
||||
|
||||
+37
-16
@@ -60,17 +60,17 @@ 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, 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, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
|
||||
JOB_ISOLATION_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
|
||||
MCP_DISABLE_TOKEN_QUERY_PARAM, MCP_DISABLE_TOKEN_QUERY_PARAM_SETTING,
|
||||
MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPMRC_SETTING, NPM_CONFIG_REGISTRY_SETTING,
|
||||
NSJAIL_TMPFS_SIZE_MB_SETTING, NSJAIL_TMP_BACKING_SETTING, NUGET_CONFIG_SETTING,
|
||||
OTEL_SETTING, OTEL_TRACES_RETENTION_SECS_SETTING, OTEL_TRACING_PROXY_SETTING,
|
||||
PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING,
|
||||
PREVIEW_TAGS_OVERRIDE_SETTING, REQUEST_SIZE_LIMIT_SETTING,
|
||||
DEFAULT_TAGS_WORKSPACES_SETTING, DEPENDENCY_JOB_TAG_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,
|
||||
JOB_DEFAULT_TIMEOUT_SECS_SETTING, JOB_ISOLATION_SETTING, JWT_SECRET_SETTING,
|
||||
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MCP_DISABLE_TOKEN_QUERY_PARAM,
|
||||
MCP_DISABLE_TOKEN_QUERY_PARAM_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPMRC_SETTING,
|
||||
NPM_CONFIG_REGISTRY_SETTING, NSJAIL_TMPFS_SIZE_MB_SETTING, NSJAIL_TMP_BACKING_SETTING,
|
||||
NUGET_CONFIG_SETTING, OTEL_SETTING, OTEL_TRACES_RETENTION_SECS_SETTING,
|
||||
OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING,
|
||||
POWERSHELL_REPO_URL_SETTING, PREVIEW_TAGS_OVERRIDE_SETTING, REQUEST_SIZE_LIMIT_SETTING,
|
||||
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
|
||||
SAML_METADATA_SETTING, SANDBOX_IMAGE_CACHE_MAX_MB_SETTING,
|
||||
SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING, SANDBOX_IMAGE_MAX_SIZE_MB_SETTING,
|
||||
@@ -93,11 +93,11 @@ use windmill_common::{
|
||||
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, 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, WORKSPACE_MAX_QUEUED_JOBS,
|
||||
WORKSPACE_MAX_QUEUED_JOBS_DEFAULT,
|
||||
DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, DEPENDENCY_JOB_TAG,
|
||||
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,
|
||||
WORKSPACE_MAX_QUEUED_JOBS, WORKSPACE_MAX_QUEUED_JOBS_DEFAULT,
|
||||
},
|
||||
KillpillSender, AUDIT_LOG_RETENTION_DAYS, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE,
|
||||
CRITICAL_ALERTS_ON_TOKEN_EXPIRY, CRITICAL_ALERT_MUTE_UI_ENABLED,
|
||||
@@ -312,6 +312,9 @@ pub async fn initial_load(
|
||||
pass.setting(PREVIEW_TAGS_OVERRIDE_SETTING, false, |v| async move {
|
||||
apply_preview_tags_override(v)
|
||||
});
|
||||
pass.setting(DEPENDENCY_JOB_TAG_SETTING, false, |v| async move {
|
||||
apply_dependency_job_tag(v)
|
||||
});
|
||||
|
||||
// Load per-workspace retention overrides before the first cleanup tick so a fresh server
|
||||
// never sweeps globally without honoring configured longer-retention workspaces.
|
||||
@@ -876,6 +879,24 @@ pub fn apply_preview_tags_override(value: Option<serde_json::Value>) {
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn load_dependency_job_tag(db: &DB) -> error::Result<()> {
|
||||
let v = load_value_from_global_settings(db, DEPENDENCY_JOB_TAG_SETTING).await?;
|
||||
apply_dependency_job_tag(v);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn apply_dependency_job_tag(value: Option<serde_json::Value>) {
|
||||
let tag = match value {
|
||||
Some(serde_json::Value::String(t)) if !t.trim().is_empty() => Some(t.trim().to_string()),
|
||||
None | Some(serde_json::Value::Null) | Some(serde_json::Value::String(_)) => None,
|
||||
Some(other) => {
|
||||
tracing::error!("{DEPENDENCY_JOB_TAG_SETTING} is not a string: {other}, ignoring");
|
||||
return;
|
||||
}
|
||||
};
|
||||
DEPENDENCY_JOB_TAG.store(std::sync::Arc::new(tag));
|
||||
}
|
||||
|
||||
// Upper bound on the duration window. Postgres `make_interval(secs => $1::int4)` is the consumer
|
||||
// downstream, so this stays comfortably below `i32::MAX` and the subsequent `u32 -> i32` cast in
|
||||
// `workspace_fairness::refresh_overloaded` cannot wrap into a negative interval (which would
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
//! Sets the process-wide `DEPENDENCY_JOB_TAG`, so it lives in its own test binary: any other
|
||||
//! test queuing a dependency job in the same process would land on the overridden tag.
|
||||
|
||||
use sqlx::{Pool, Postgres};
|
||||
use windmill_common::worker::DEPENDENCY_JOB_TAG;
|
||||
use windmill_dep_map::scoped_dependency_map::DependencyDependent;
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn relocks_go_to_the_configured_dependency_job_tag(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
DEPENDENCY_JOB_TAG.store(std::sync::Arc::new(Some("deps-$workspace".to_string())));
|
||||
|
||||
for (path, hash, tag) in [
|
||||
("f/tags/tagged", 7_100_001_i64, Some("default-$workspace")),
|
||||
("f/tags/plain", 7_100_002_i64, None),
|
||||
] {
|
||||
sqlx::query(
|
||||
"INSERT INTO script(workspace_id, created_by, content, schema, summary, description, path, hash, language, tag)
|
||||
VALUES ('test-workspace', 'test-user', 'def main(): pass', '{}', '', '', $1, $2, 'python3', $3)",
|
||||
)
|
||||
.bind(path)
|
||||
.bind(hash)
|
||||
.bind(tag)
|
||||
.execute(&db)
|
||||
.await?;
|
||||
}
|
||||
let importers = ["f/tags/tagged", "f/tags/plain"]
|
||||
.map(|p| DependencyDependent {
|
||||
importer_path: p.to_string(),
|
||||
importer_kind: "script".to_string(),
|
||||
importer_node_ids: None,
|
||||
})
|
||||
.to_vec();
|
||||
windmill_dep_map::trigger_dependents::trigger_dependents_to_recompute_dependencies(
|
||||
"test-workspace",
|
||||
importers,
|
||||
None,
|
||||
None,
|
||||
"test@windmill.dev",
|
||||
"test-user",
|
||||
"u/test-user",
|
||||
&db,
|
||||
vec![],
|
||||
)
|
||||
.await?;
|
||||
|
||||
let tags: Vec<(String, String)> = sqlx::query_as(
|
||||
"SELECT runnable_path, tag FROM v2_job WHERE kind = 'dependencies' ORDER BY runnable_path",
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.await?;
|
||||
assert_eq!(
|
||||
tags,
|
||||
vec![
|
||||
(
|
||||
"f/tags/plain".to_string(),
|
||||
"deps-test-workspace".to_string()
|
||||
),
|
||||
(
|
||||
"f/tags/tagged".to_string(),
|
||||
"deps-test-workspace".to_string()
|
||||
),
|
||||
]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
@@ -9,6 +9,7 @@ pub const DEFAULT_TAGS_WORKSPACES_SETTING: &str = "default_tags_workspaces";
|
||||
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 BASE_URL_SETTING: &str = "base_url";
|
||||
pub const WS_BASE_URL_SETTING: &str = "ws_base_url";
|
||||
pub const OAUTH_SETTING: &str = "oauths";
|
||||
|
||||
@@ -275,6 +275,8 @@ pub struct GlobalSettings {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub auto_build_binary_tag: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub dependency_job_tag: 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>,
|
||||
|
||||
@@ -28,7 +28,9 @@ use crate::utils::StripPath;
|
||||
/// `push`'s default (the language tag). Every relock of a script must use this, or a relock
|
||||
/// lands on a worker pool other than the one its deploy locked on and resolves packages
|
||||
/// differently.
|
||||
/// Dedicated workers are handled by `push`, which gives `dedicated_worker` precedence.
|
||||
/// Dedicated workers are handled by `push`, which gives `dedicated_worker` precedence. Below
|
||||
/// that, `push` replaces this result with the instance's `dependency_job_tag` setting when it is
|
||||
/// set, except for bunnative.
|
||||
pub fn dependency_job_tag(tag: Option<String>, language: &ScriptLang) -> Option<String> {
|
||||
if tag.as_ref().is_some_and(|x| x.contains("$args[")) {
|
||||
None
|
||||
|
||||
@@ -450,6 +450,7 @@ lazy_static::lazy_static! {
|
||||
pub static ref DEFAULT_TAGS_WORKSPACES: arc_swap::ArcSwap<Option<Vec<String>>> = arc_swap::ArcSwap::from_pointee(None);
|
||||
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);
|
||||
|
||||
pub static ref MAX_TIMEOUT: u64 = std::env::var("TIMEOUT")
|
||||
.ok()
|
||||
|
||||
@@ -81,9 +81,9 @@ use windmill_common::{
|
||||
users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL},
|
||||
utils::{not_found_if_none, report_critical_error, StripPath, WarnAfterExt},
|
||||
worker::{
|
||||
to_raw_value, CLOUD_HOSTED, DISABLE_FLOW_SCRIPT, NO_LOGS, PREVIEW_TAGS_OVERRIDE,
|
||||
WORKER_PULL_QUERIES, WORKER_PULL_QUERIES_FAIRNESS, WORKER_SUSPENDED_PULL_QUERY,
|
||||
WORKSPACE_FAIRNESS_OVERLOADED,
|
||||
to_raw_value, CLOUD_HOSTED, DEPENDENCY_JOB_TAG, DISABLE_FLOW_SCRIPT, NO_LOGS,
|
||||
PREVIEW_TAGS_OVERRIDE, WORKER_PULL_QUERIES, WORKER_PULL_QUERIES_FAIRNESS,
|
||||
WORKER_SUSPENDED_PULL_QUERY, WORKSPACE_FAIRNESS_OVERLOADED,
|
||||
},
|
||||
DB, METRICS_ENABLED,
|
||||
};
|
||||
@@ -4619,15 +4619,27 @@ pub async fn pull(
|
||||
&& !(job.kind.is_preview()
|
||||
&& PREVIEW_TAGS_OVERRIDE.load(std::sync::atomic::Ordering::Relaxed))
|
||||
{
|
||||
let effective_ws = per_workspace_tag(&job.workspace_id, db).await;
|
||||
let base_tag = if job.is_flow() {
|
||||
"flow".to_string()
|
||||
} else {
|
||||
"dependency".to_string()
|
||||
};
|
||||
let tag = match &effective_ws {
|
||||
Some(ws) => format!("{}-{}", base_tag, ws),
|
||||
None => base_tag,
|
||||
let tag = match configured_dependency_job_tag(
|
||||
&job.kind,
|
||||
job.script_lang.as_ref(),
|
||||
&job.workspace_id,
|
||||
db,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Some(tag) => tag,
|
||||
None => {
|
||||
let effective_ws = per_workspace_tag(&job.workspace_id, db).await;
|
||||
let base_tag = if job.is_flow() {
|
||||
"flow".to_string()
|
||||
} else {
|
||||
"dependency".to_string()
|
||||
};
|
||||
match &effective_ws {
|
||||
Some(ws) => format!("{}-{}", base_tag, ws),
|
||||
None => base_tag,
|
||||
}
|
||||
}
|
||||
};
|
||||
sqlx::query!(
|
||||
"UPDATE v2_job_queue SET tag = $1, running = false WHERE id = $2",
|
||||
@@ -5275,6 +5287,24 @@ pub async fn resolve_push_tag(
|
||||
Some(interpolate_args(tag.to_string(), args, &tag_ws))
|
||||
}
|
||||
|
||||
/// The tag a dependency job is routed to when the instance's `dependency_job_tag` setting is
|
||||
/// set, whatever tag the script, flow or app itself runs on. `None` for any other job kind, for
|
||||
/// bunnative (its bundle must be built on a worker with the bun tag), or when the setting is
|
||||
/// unset.
|
||||
pub async fn configured_dependency_job_tag(
|
||||
job_kind: &JobKind,
|
||||
language: Option<&ScriptLang>,
|
||||
workspace_id: &str,
|
||||
db: &DB,
|
||||
) -> Option<String> {
|
||||
if !job_kind.is_dependency() || language == Some(&ScriptLang::Bunnative) {
|
||||
return None;
|
||||
}
|
||||
let tag = DEPENDENCY_JOB_TAG.load_full();
|
||||
let tag = tag.as_deref()?;
|
||||
resolve_push_tag(tag, &PushArgs::from(&HashMap::new()), workspace_id, db).await
|
||||
}
|
||||
|
||||
/// Refuses a `tag` the caller chose that the instance's custom tags do not let `w_id` use,
|
||||
/// judging the queue it resolves to. `args` must be the ones the job is pushed with: resolving
|
||||
/// with any others checks a queue the job does not land on.
|
||||
@@ -7152,6 +7182,10 @@ async fn push_inner<'c, 'd>(
|
||||
runnable_path.clone().expect("dedicated script has a path")
|
||||
);
|
||||
windmill_common::worker::dedicated_worker_tag(workspace_id, &full_path)
|
||||
} else if let Some(tag) =
|
||||
configured_dependency_job_tag(&job_kind, language.as_ref(), workspace_id, db).await
|
||||
{
|
||||
tag
|
||||
} else {
|
||||
let interpolated_tag = match tag {
|
||||
Some(x) => resolve_push_tag(&x, &args, workspace_id, db).await,
|
||||
|
||||
@@ -8,8 +8,10 @@
|
||||
DEFAULT_TAGS_PER_WORKSPACE_SETTING,
|
||||
DEFAULT_TAGS_WORKSPACES_SETTING,
|
||||
FORK_WORKSPACE_TAG_APPEND_FORK_SUFFIX_SETTING,
|
||||
PREVIEW_TAGS_OVERRIDE_SETTING
|
||||
PREVIEW_TAGS_OVERRIDE_SETTING,
|
||||
DEPENDENCY_JOB_TAG_SETTING
|
||||
} from '$lib/consts'
|
||||
import TextInput from './text_input/TextInput.svelte'
|
||||
import Toggle from './Toggle.svelte'
|
||||
import MultiSelect from './select/MultiSelect.svelte'
|
||||
import { safeSelectItems } from './select/utils.svelte'
|
||||
@@ -31,12 +33,14 @@
|
||||
let limitToWorkspaces = $state(false)
|
||||
let previewTagsOverride = $state(false)
|
||||
let forkAppendForkSuffix = $state(false)
|
||||
let dependencyJobTag = $state('')
|
||||
|
||||
// Change detection
|
||||
let originalDefaultTagPerWorkspace = $state<boolean | undefined>(defaultTagPerWorkspace)
|
||||
let originalDefaultTagWorkspaces = $state<string[]>(defaultTagWorkspaces)
|
||||
let originalPreviewTagsOverride = $state(false)
|
||||
let originalForkAppendForkSuffix = $state(false)
|
||||
let originalDependencyJobTag = $state('')
|
||||
|
||||
// Detect changes
|
||||
let hasChanges = $derived(
|
||||
@@ -44,7 +48,8 @@
|
||||
JSON.stringify($state.snapshot(originalDefaultTagWorkspaces)?.sort() || []) !==
|
||||
JSON.stringify($state.snapshot(defaultTagWorkspaces)?.sort() || []) ||
|
||||
originalPreviewTagsOverride !== previewTagsOverride ||
|
||||
originalForkAppendForkSuffix !== forkAppendForkSuffix
|
||||
originalForkAppendForkSuffix !== forkAppendForkSuffix ||
|
||||
originalDependencyJobTag.trim() !== dependencyJobTag.trim()
|
||||
)
|
||||
|
||||
let workspaces: string[] = $state([])
|
||||
@@ -70,6 +75,11 @@
|
||||
})) as any
|
||||
forkAppendForkSuffix = forkSetting ?? false
|
||||
originalForkAppendForkSuffix = forkAppendForkSuffix
|
||||
dependencyJobTag =
|
||||
((await SettingService.getGlobal({
|
||||
key: DEPENDENCY_JOB_TAG_SETTING
|
||||
})) as any) ?? ''
|
||||
originalDependencyJobTag = dependencyJobTag
|
||||
} catch (err) {
|
||||
sendUserToast(`Could not load default tags: ${err}`, true)
|
||||
}
|
||||
@@ -104,11 +114,19 @@
|
||||
}
|
||||
})
|
||||
|
||||
await SettingService.setGlobal({
|
||||
key: DEPENDENCY_JOB_TAG_SETTING,
|
||||
requestBody: {
|
||||
value: dependencyJobTag.trim() || undefined
|
||||
}
|
||||
})
|
||||
|
||||
// Update original state after save
|
||||
originalDefaultTagPerWorkspace = defaultTagPerWorkspace
|
||||
originalDefaultTagWorkspaces = [...(defaultTagWorkspaces || [])]
|
||||
originalPreviewTagsOverride = previewTagsOverride
|
||||
originalForkAppendForkSuffix = forkAppendForkSuffix
|
||||
originalDependencyJobTag = dependencyJobTag
|
||||
|
||||
loadDefaultTags()
|
||||
sendUserToast('Saved')
|
||||
@@ -218,6 +236,33 @@
|
||||
disabled={!$enterpriseLicense}
|
||||
/>
|
||||
</div>
|
||||
<div class="flex flex-col gap-1 max-w-md">
|
||||
<span class="text-xs font-semibold text-emphasis">Dependency job tag</span>
|
||||
<span class="text-2xs text-secondary">
|
||||
Route every dependency job (lockfile resolution at deploy and relocks triggered by an
|
||||
imported script changing, for scripts, flows and apps) to this tag, whatever tag the
|
||||
runnable itself runs on. Supports <code>$workspace</code>. Dedicated workers still build
|
||||
their own locks, and Bun native scripts still build their bundle on the bun tag.
|
||||
</span>
|
||||
<span class="text-2xs text-secondary">
|
||||
Leave empty for the default routing: a script's dependency job runs on the script's own
|
||||
tag, or on its language's default tag (e.g. <code
|
||||
>{defaultTagPerWorkspace ? 'python3-$workspace' : 'python3'}</code
|
||||
>) when it has none; flow and app dependency jobs run on
|
||||
<code>{defaultTagPerWorkspace ? 'dependency-$workspace' : 'dependency'}</code>.
|
||||
<a
|
||||
href="https://www.windmill.dev/docs/core_concepts/worker_groups"
|
||||
target="_blank"
|
||||
class="gap-1 items-baseline"
|
||||
>Worker groups <ExternalLink size={12} class="inline-block" /></a
|
||||
>
|
||||
</span>
|
||||
<TextInput
|
||||
bind:value={dependencyJobTag}
|
||||
size="sm"
|
||||
inputProps={{ placeholder: 'e.g. dependency', disabled: !$enterpriseLicense }}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="flex gap-2 items-center mb-1">
|
||||
@@ -240,6 +285,17 @@
|
||||
</div>
|
||||
</div>
|
||||
{/each}
|
||||
{#if dependencyJobTag.trim()}
|
||||
<div class="flex gap-2 items-center">
|
||||
<div class="w-36">
|
||||
<Badge color="transparent">all dependency jobs</Badge>
|
||||
</div>
|
||||
<div class="w-6 flex justify-center text-secondary">→</div>
|
||||
<div class="flex-1">
|
||||
<Badge color="blue">{dependencyJobTag.trim()}</Badge>
|
||||
</div>
|
||||
</div>
|
||||
{/if}
|
||||
{#if previewTagsOverride}
|
||||
<div class="flex gap-2 items-center">
|
||||
<div class="w-36">
|
||||
|
||||
@@ -47,6 +47,7 @@ export const DEFAULT_TAGS_PER_WORKSPACE_SETTING = 'default_tags_per_workspace'
|
||||
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 WORKSPACE_FAIRNESS_ENABLED_SETTING = 'workspace_fairness_enabled'
|
||||
export const WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING = 'workspace_fairness_max_percent'
|
||||
|
||||
Reference in New Issue
Block a user