mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-12 08:02:44 +00:00
feat: pause a worker group without stopping its workers (#11596)
* feat: pause a worker group without stopping its workers * chore: bump ee ref * fix: keep dedicated workers running across a pause toggle * fix: edit a copy of the worker group config in the drawer * fix: stable priority tag order, keep pause-only configs in the drawer * fix: keep inherited worker tags when saving from the drawer * fix: keep an explicit empty tag list when leaving dedicated mode * fix: keep inherited tags when creating a config from the drawer Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * chore: update ee-repo-ref to 6f23819e947b18878000e580ddefccfd01a73166 This commit updates the EE repository reference after PR #853 was merged in windmill-ee-private. Previous ee-repo-ref: 6cf802bdbe17b4a66757016515044b8a9ef9ef19 New ee-repo-ref: 6f23819e947b18878000e580ddefccfd01a73166 Automated by sync-ee-ref workflow. * feat: confirm before applying a worker config change that restarts workers Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * chore: pin the ee commit that compiles against main Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * chore: retrigger ci after the github outage dropped the push event Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <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
14381cefee
commit
95c4b3096c
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "coalesce",
|
||||
"type_info": "Bool"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "91fdd565bd210f8aa4c6d5f5e309573aefaeb5dbc6ac07f5e664bc8fe2cb2683"
|
||||
}
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "coalesce",
|
||||
"type_info": "Bool"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "a6d355b3121f199f3827c05aaa7c8f780ab4c9d3d373d3cd0bb489e0d173040c"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
0b78e2c6fefcbad3c465bf75ebfd6ebd6912ad19
|
||||
1600d49ad5513af8b54ad65688d3d1c7edc977d9
|
||||
@@ -427,6 +427,8 @@ pub async fn initial_load(
|
||||
native_mode,
|
||||
// an agent worker never reads its group's config, only its token
|
||||
object_store_cache_config: None,
|
||||
// nor whether it is paused: the server refuses it jobs instead
|
||||
paused: false,
|
||||
}));
|
||||
}
|
||||
}
|
||||
@@ -5689,8 +5691,13 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b
|
||||
.as_ref()
|
||||
.is_some_and(|dws| !dws.is_empty());
|
||||
|
||||
// Pausing or resuming must not restart the workers, dedicated ones included: a reload
|
||||
// whose only difference is the flag skips every restart below.
|
||||
let only_pause_changed = wc.paused != config.paused
|
||||
&& WorkerConfig { paused: wc.paused, ..config.clone() } == **wc;
|
||||
|
||||
if **wc != config || has_dedicated {
|
||||
if kill_if_change {
|
||||
if kill_if_change && !only_pause_changed {
|
||||
if has_dedicated
|
||||
|| wc.dedicated_worker != config.dedicated_worker
|
||||
|| wc.dedicated_workers != config.dedicated_workers
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
//! In its own test binary: it flips the process-wide `WORKER_CONFIG`, which would stall every
|
||||
//! other worker test sharing the process.
|
||||
|
||||
use futures::StreamExt;
|
||||
use sqlx::{Pool, Postgres};
|
||||
use std::time::Duration;
|
||||
use windmill_common::{
|
||||
jobs::{JobPayload, RawCode},
|
||||
scripts::ScriptLang,
|
||||
worker::{Connection, WORKER_CONFIG},
|
||||
};
|
||||
use windmill_test_utils::*;
|
||||
|
||||
fn set_paused(paused: bool) {
|
||||
let mut wc = (**WORKER_CONFIG.load()).clone();
|
||||
wc.paused = paused;
|
||||
WORKER_CONFIG.store(std::sync::Arc::new(wc));
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn paused_worker_leaves_jobs_queued_until_resumed(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
initialize_tracing().await;
|
||||
let server = ApiServer::start(db.clone()).await?;
|
||||
let port = server.addr.port();
|
||||
|
||||
set_paused(true);
|
||||
let id = RunJob::from(JobPayload::Code(RawCode {
|
||||
hash: None,
|
||||
content: "echo done".to_string(),
|
||||
path: None,
|
||||
lock: None,
|
||||
language: ScriptLang::Bash,
|
||||
cache_ttl: None,
|
||||
cache_ignore_s3_path: None,
|
||||
dedicated_worker: None,
|
||||
concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default()
|
||||
.into(),
|
||||
debouncing_settings: windmill_common::runnable_settings::DebouncingSettings::default(),
|
||||
modules: None,
|
||||
tag: None,
|
||||
}))
|
||||
.push(&db)
|
||||
.await;
|
||||
|
||||
let mut completed = listen_for_completed_jobs(&db).await;
|
||||
in_test_worker(
|
||||
Connection::Sql(db.clone()),
|
||||
async {
|
||||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||
let running =
|
||||
sqlx::query_scalar::<_, bool>("SELECT running FROM v2_job_queue WHERE id = $1")
|
||||
.bind(id)
|
||||
.fetch_optional(&db)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
running,
|
||||
Some(false),
|
||||
"a paused worker must not pull the job"
|
||||
);
|
||||
|
||||
set_paused(false);
|
||||
while completed.next().await != Some(id) {}
|
||||
},
|
||||
port,
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(completed_job(id, &db).await.success);
|
||||
Ok(())
|
||||
}
|
||||
@@ -224,12 +224,13 @@ async fn update_config(
|
||||
|
||||
#[cfg(not(feature = "enterprise"))]
|
||||
let config = if name.starts_with("worker__") {
|
||||
// In CE, only allow setting worker_tags, cache_clear, init_bash, and native_mode
|
||||
// In CE, only allow setting worker_tags, cache_clear, init_bash, native_mode and paused
|
||||
serde_json::json!({
|
||||
"worker_tags": config.get("worker_tags"),
|
||||
"cache_clear": config.get("cache_clear"),
|
||||
"init_bash": config.get("init_bash"),
|
||||
"native_mode": config.get("native_mode")
|
||||
"native_mode": config.get("native_mode"),
|
||||
"paused": config.get("paused")
|
||||
})
|
||||
} else {
|
||||
config
|
||||
@@ -272,6 +273,18 @@ async fn update_config(
|
||||
}
|
||||
|
||||
let mut tx = db.begin().await?;
|
||||
let was_paused = if name.starts_with("worker__") {
|
||||
sqlx::query_scalar!(
|
||||
"SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1 FOR UPDATE",
|
||||
&name
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten()
|
||||
.unwrap_or(false)
|
||||
} else {
|
||||
false
|
||||
};
|
||||
sqlx::query!(
|
||||
"INSERT INTO config (name, config) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET config = EXCLUDED.config",
|
||||
&name,
|
||||
@@ -291,6 +304,17 @@ async fn update_config(
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
|
||||
if name.starts_with("worker__") {
|
||||
let paused = config.get("paused").and_then(|v| v.as_bool()) == Some(true);
|
||||
if paused != was_paused {
|
||||
windmill_common::feature_usage::log_feature_usage(
|
||||
"worker_group",
|
||||
"pause_toggle",
|
||||
if paused { "pause" } else { "resume" },
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(format!("Updated config {name}"))
|
||||
}
|
||||
|
||||
|
||||
@@ -963,6 +963,9 @@ pub struct WorkerGroupConfig {
|
||||
pub autoscaling: Option<AutoscalingConfig>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub native_mode: Option<bool>,
|
||||
/// 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>,
|
||||
/// 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")]
|
||||
|
||||
@@ -477,6 +477,7 @@ lazy_static::lazy_static! {
|
||||
env_vars: Default::default(),
|
||||
native_mode: false,
|
||||
object_store_cache_config: Default::default(),
|
||||
paused: false,
|
||||
});
|
||||
|
||||
pub static ref WORKER_PULL_QUERIES: arc_swap::ArcSwap<Vec<String>> = arc_swap::ArcSwap::from_pointee(vec![]);
|
||||
@@ -844,11 +845,14 @@ pub fn make_suspended_pull_query(tags: &[String]) -> String {
|
||||
}
|
||||
// pub async fn make_suspended
|
||||
pub async fn store_suspended_pull_query(wc: &WorkerConfig) {
|
||||
if wc.worker_tags.len() == 0 {
|
||||
tracing::error!("Empty tags in worker tags, skipping");
|
||||
return;
|
||||
}
|
||||
let query = make_suspended_pull_query(&wc.worker_tags);
|
||||
// An empty tag list must clear the query: keeping the previous one would leave a reloaded
|
||||
// worker pulling suspended jobs for tags it no longer listens to.
|
||||
let query = if wc.worker_tags.is_empty() {
|
||||
tracing::error!("Empty tags in worker tags, pulling no suspended jobs");
|
||||
String::new()
|
||||
} else {
|
||||
make_suspended_pull_query(&wc.worker_tags)
|
||||
};
|
||||
WORKER_SUSPENDED_PULL_QUERY.store(std::sync::Arc::new(query));
|
||||
}
|
||||
|
||||
@@ -922,6 +926,42 @@ pub async fn store_pull_query(wc: &WorkerConfig) {
|
||||
WORKER_PULL_QUERIES_FAIRNESS.store(std::sync::Arc::new(fairness_queries));
|
||||
}
|
||||
|
||||
const WORKER_GROUP_PAUSED_CACHE_TTL: Duration = Duration::from_secs(10);
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref WORKER_GROUP_PAUSED_CACHE: quick_cache::sync::Cache<String, (bool, std::time::Instant)> =
|
||||
quick_cache::sync::Cache::new(1000);
|
||||
}
|
||||
|
||||
/// Whether the config of `worker_group` is paused, as the server sees it. A worker that reads
|
||||
/// its own config knows from `WORKER_CONFIG`; this is for the server deciding on behalf of
|
||||
/// workers that do not, and is cached since they ask on every pull.
|
||||
pub async fn is_worker_group_paused(db: &DB, worker_group: &str) -> bool {
|
||||
if let Some((paused, at)) = WORKER_GROUP_PAUSED_CACHE.get(worker_group) {
|
||||
if at.elapsed() < WORKER_GROUP_PAUSED_CACHE_TTL {
|
||||
return paused;
|
||||
}
|
||||
}
|
||||
let paused = match sqlx::query_scalar!(
|
||||
"SELECT COALESCE(config->'paused' = 'true'::jsonb, false) FROM config WHERE name = $1",
|
||||
format!("worker__{worker_group}")
|
||||
)
|
||||
.fetch_optional(db)
|
||||
.await
|
||||
{
|
||||
Ok(paused) => paused.flatten().unwrap_or(false),
|
||||
Err(e) => {
|
||||
tracing::error!("Could not read whether worker group {worker_group} is paused: {e:#}");
|
||||
return false;
|
||||
}
|
||||
};
|
||||
WORKER_GROUP_PAUSED_CACHE.insert(
|
||||
worker_group.to_string(),
|
||||
(paused, std::time::Instant::now()),
|
||||
);
|
||||
paused
|
||||
}
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
pub static ref WINDMILL_DIR: String = {
|
||||
let dir = std::env::var("WINDMILL_DIR")
|
||||
@@ -2470,6 +2510,12 @@ pub async fn load_worker_config(
|
||||
.collect();
|
||||
priority_tags_sorted.push(PriorityTags { priority: 0, tags: Vec::from_iter(all_tags_set) }); // push the tags that were not listed as high priority with a priority = 0
|
||||
priority_tags_sorted.sort_by_key(|elt| Reverse(elt.priority)); // sort by priority DESC
|
||||
|
||||
// The buckets come out of hash maps: without a fixed order two loads of the same
|
||||
// config compare unequal, and a reload reads that as a config change.
|
||||
for bucket in priority_tags_sorted.iter_mut() {
|
||||
bucket.tags.sort();
|
||||
}
|
||||
} else {
|
||||
// if no priority is used, push all tags with a priority to 0
|
||||
priority_tags_sorted.push(PriorityTags { priority: 0, tags: worker_tags.clone() });
|
||||
@@ -2539,6 +2585,7 @@ pub async fn load_worker_config(
|
||||
env_vars: resolved_env_vars,
|
||||
native_mode,
|
||||
object_store_cache_config: config.object_store_cache_config,
|
||||
paused: config.paused.unwrap_or(false),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -2629,6 +2676,7 @@ pub struct WorkerConfigOpt {
|
||||
pub env_vars_allowlist: Option<Vec<String>>,
|
||||
pub native_mode: Option<bool>,
|
||||
pub object_store_cache_config: Option<serde_json::Value>,
|
||||
pub paused: Option<bool>,
|
||||
}
|
||||
|
||||
impl Default for WorkerConfigOpt {
|
||||
@@ -2648,6 +2696,7 @@ impl Default for WorkerConfigOpt {
|
||||
env_vars_allowlist: Default::default(),
|
||||
native_mode: Default::default(),
|
||||
object_store_cache_config: Default::default(),
|
||||
paused: Default::default(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2670,12 +2719,15 @@ pub struct WorkerConfig {
|
||||
/// in the group config. Raw JSON: `windmill-common` cannot depend on the object store crate
|
||||
/// that parses it, and comparing the raw value is what tells a reload the store changed.
|
||||
pub object_store_cache_config: Option<serde_json::Value>,
|
||||
/// The group pulls no job from the queue: its workers finish what they hold, keep pinging
|
||||
/// and leave queued jobs for when it is resumed.
|
||||
pub paused: bool,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for WorkerConfig {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, object_store_cache_config: {} }}",
|
||||
self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "), self.native_mode,
|
||||
write!(f, "WorkerConfig {{ worker_tags: {:?}, priority_tags_sorted: {:?}, dedicated_worker: {:?}, dedicated_workers: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?}, native_mode: {:?}, paused: {:?}, object_store_cache_config: {} }}",
|
||||
self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, self.dedicated_workers, self.init_bash, self.periodic_script_bash, self.periodic_script_interval_seconds, self.cache_clear, self.additional_python_paths, self.pip_local_dependencies, self.env_vars.iter().map(|(k, v)| format!("{}: {}{} ({} chars)", k, &v[..3.min(v.len())], "***", v.len())).collect::<Vec<String>>().join(", "), self.native_mode, self.paused,
|
||||
// holds bucket credentials
|
||||
self.object_store_cache_config.as_ref().map(|_| "***").unwrap_or("None"))
|
||||
}
|
||||
|
||||
@@ -3424,6 +3424,10 @@ pub async fn run_worker(
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
continue;
|
||||
}
|
||||
} else if WORKER_CONFIG.load().paused {
|
||||
// Checked after the same-worker channel: a flow this worker already runs
|
||||
// still needs it for its remaining same-worker steps.
|
||||
Ok(None)
|
||||
} else {
|
||||
match &conn {
|
||||
Connection::Sql(db) => {
|
||||
|
||||
@@ -104,6 +104,7 @@
|
||||
autoscaling?: AutoscalingConfig
|
||||
native_mode?: boolean
|
||||
object_store_cache_config?: ObjectStoreConfig
|
||||
paused?: boolean
|
||||
} = $state({})
|
||||
|
||||
function loadNConfig() {
|
||||
@@ -111,10 +112,9 @@
|
||||
? config.worker_tags != undefined ||
|
||||
config.dedicated_worker != undefined ||
|
||||
config.dedicated_workers != undefined
|
||||
? config
|
||||
: {
|
||||
worker_tags: []
|
||||
}
|
||||
? // a copy: editing the draft must not touch the saved config it is compared against
|
||||
$state.snapshot(config)
|
||||
: { ...$state.snapshot(config), worker_tags: [] }
|
||||
: {
|
||||
worker_tags: []
|
||||
}
|
||||
@@ -212,6 +212,7 @@
|
||||
periodic_script_interval_seconds?: number
|
||||
native_mode?: boolean
|
||||
object_store_cache_config?: ObjectStoreConfig
|
||||
paused?: boolean
|
||||
}
|
||||
activeWorkers: number
|
||||
customTags: string[] | undefined
|
||||
@@ -268,8 +269,10 @@
|
||||
if (!config && !nconfig) return false
|
||||
if (!config || !nconfig) return true
|
||||
|
||||
const cleaned1 = cleanValueProperties(config)
|
||||
const cleaned2 = cleanValueProperties(nconfig)
|
||||
// Compared without empty values: the draft is filled with empty defaults a saved
|
||||
// config may lack, which are not an edit.
|
||||
const cleaned1 = cleanWorkerGroupConfig(cleanValueProperties(config))
|
||||
const cleaned2 = cleanWorkerGroupConfig(cleanValueProperties(nconfig))
|
||||
|
||||
return (
|
||||
orderedJsonStringify(replaceFalseWithUndefined(cleaned1)) !==
|
||||
@@ -298,6 +301,47 @@
|
||||
let openDelete = $state(false)
|
||||
let openClean = $state(false)
|
||||
let openRestart = $state(false)
|
||||
let pendingConfig: typeof nconfig | undefined = $state()
|
||||
|
||||
// Mirrors `reload_worker_config` in backend/src/monitor.rs, which decides the restart. It
|
||||
// compares the body that is sent rather than the draft, and only the fields a worker loads.
|
||||
const RESTART_FIELDS = [
|
||||
'dedicated_worker',
|
||||
'dedicated_workers',
|
||||
'init_bash',
|
||||
'periodic_script_bash',
|
||||
'periodic_script_interval_seconds',
|
||||
'native_mode'
|
||||
] as const
|
||||
const OTHER_WORKER_FIELDS = [
|
||||
'worker_tags',
|
||||
'priority_tags',
|
||||
'additional_python_paths',
|
||||
'pip_local_dependencies',
|
||||
'object_store_cache_config'
|
||||
] as const
|
||||
function restartsWorkers(body: typeof nconfig): boolean {
|
||||
if (activeWorkers === 0) return false
|
||||
const saved: typeof nconfig = config ?? {}
|
||||
const comparable = (c: typeof nconfig, k: keyof typeof nconfig) =>
|
||||
orderedJsonStringify(replaceFalseWithUndefined(cleanWorkerGroupConfig({ v: c[k] })))
|
||||
const changed = (k: keyof typeof nconfig) => comparable(saved, k) !== comparable(body, k)
|
||||
const restartFieldChanged = RESTART_FIELDS.some(changed)
|
||||
const onlyPauseChanged =
|
||||
changed('paused') &&
|
||||
!restartFieldChanged &&
|
||||
!OTHER_WORKER_FIELDS.some(changed) &&
|
||||
!hasEnvVarChanges
|
||||
const dedicated =
|
||||
body.dedicated_worker != undefined || (body.dedicated_workers?.length ?? 0) > 0
|
||||
return !onlyPauseChanged && (dedicated || restartFieldChanged)
|
||||
}
|
||||
|
||||
async function saveConfig(body: typeof nconfig) {
|
||||
await ConfigService.updateConfig({ name: 'worker__' + name, requestBody: body })
|
||||
sendUserToast('Configuration set')
|
||||
dispatch('reload')
|
||||
}
|
||||
|
||||
// Compute hashed tags for display (actual tags used by the worker)
|
||||
let hashedDedicatedTags: Map<string, string> = $state(new Map())
|
||||
@@ -399,6 +443,31 @@
|
||||
</div>
|
||||
</ConfirmationModal>
|
||||
|
||||
<ConfirmationModal
|
||||
open={pendingConfig != undefined}
|
||||
title="Apply changes and restart workers"
|
||||
confirmationText="Apply and restart"
|
||||
on:canceled={() => {
|
||||
pendingConfig = undefined
|
||||
}}
|
||||
on:confirmed={async () => {
|
||||
const body = pendingConfig
|
||||
pendingConfig = undefined
|
||||
if (body) {
|
||||
await saveConfig(body)
|
||||
}
|
||||
}}
|
||||
>
|
||||
<div class="flex flex-col w-full space-y-4">
|
||||
<span
|
||||
>Applying these changes will restart the {pluralize(activeWorkers, 'worker')} of worker group '{name}'.
|
||||
Workers finish their running job, shut down gracefully and are expected to be restarted by
|
||||
their supervisor.</span
|
||||
>
|
||||
<span>Pausing or resuming a worker group alone does not restart its workers.</span>
|
||||
</div>
|
||||
</ConfirmationModal>
|
||||
|
||||
<ConfirmationModal
|
||||
open={openRestart}
|
||||
title="Restart workers"
|
||||
@@ -733,6 +802,32 @@
|
||||
</div>
|
||||
{/if}
|
||||
|
||||
{#if nconfig !== undefined}
|
||||
<div class="mt-8"></div>
|
||||
<Label label="Pause">
|
||||
{#snippet header()}
|
||||
<Tooltip>
|
||||
{#snippet text()}
|
||||
A paused worker group stops pulling jobs, including suspended jobs that are resumed.
|
||||
Jobs already running finish, queued jobs stay in the queue until the group is resumed.
|
||||
Workers pick the change up within about 10 seconds, without restarting.
|
||||
{/snippet}
|
||||
</Tooltip>
|
||||
{/snippet}
|
||||
<Toggle
|
||||
size="sm"
|
||||
options={{ right: 'Pause this worker group' }}
|
||||
checked={nconfig?.paused === true}
|
||||
on:change={(ev) => {
|
||||
if (nconfig !== undefined) {
|
||||
nconfig.paused = ev.detail ? true : undefined
|
||||
}
|
||||
}}
|
||||
disabled={!canEditConfig}
|
||||
/>
|
||||
</Label>
|
||||
{/if}
|
||||
|
||||
<div class="mt-8"></div>
|
||||
|
||||
<Label
|
||||
@@ -1216,9 +1311,19 @@
|
||||
}
|
||||
})
|
||||
|
||||
await ConfigService.updateConfig({ name: 'worker__' + name, requestBody: nconfig })
|
||||
sendUserToast('Configuration set')
|
||||
dispatch('reload')
|
||||
// A group without saved tags inherits them from the worker's environment, and
|
||||
// an explicit empty list would override that: only send one the user asked for.
|
||||
const inheritsTags =
|
||||
config?.worker_tags == undefined &&
|
||||
config?.dedicated_worker == undefined &&
|
||||
config?.dedicated_workers == undefined &&
|
||||
nconfig.worker_tags?.length === 0
|
||||
const body = inheritsTags ? { ...nconfig, worker_tags: undefined } : nconfig
|
||||
if (restartsWorkers(body)) {
|
||||
pendingConfig = body
|
||||
} else {
|
||||
await saveConfig(body)
|
||||
}
|
||||
}}
|
||||
disabled={(!hasChanges &&
|
||||
!hasEnvVarChanges &&
|
||||
@@ -1258,6 +1363,16 @@
|
||||
{#if isNativeMode}
|
||||
<Badge color="blue" small>Native</Badge>
|
||||
{/if}
|
||||
{#if config?.paused === true}
|
||||
<Badge color="yellow" small>
|
||||
Paused
|
||||
<Tooltip>
|
||||
{#snippet text()}
|
||||
Workers of this group pull no new jobs. Queued jobs wait until the group is resumed.
|
||||
{/snippet}
|
||||
</Tooltip>
|
||||
</Badge>
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
{#if vcpus_memory?.vcpus}
|
||||
|
||||
Reference in New Issue
Block a user