diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2e6878b034..11006d08eb 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -c05572e93739e2697ab310d87efe2744cd0e1aaf +a18ac31062ac092cb9a5fc87629e217d97f4911d diff --git a/backend/migrations/20260129161851_dedicated_workers_column.down.sql b/backend/migrations/20260129161851_dedicated_workers_column.down.sql new file mode 100644 index 0000000000..44927fbe0f --- /dev/null +++ b/backend/migrations/20260129161851_dedicated_workers_column.down.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping DROP COLUMN IF EXISTS dedicated_workers; diff --git a/backend/migrations/20260129161851_dedicated_workers_column.up.sql b/backend/migrations/20260129161851_dedicated_workers_column.up.sql new file mode 100644 index 0000000000..b42aae58a3 --- /dev/null +++ b/backend/migrations/20260129161851_dedicated_workers_column.up.sql @@ -0,0 +1 @@ +ALTER TABLE worker_ping ADD COLUMN IF NOT EXISTS dedicated_workers TEXT[]; diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 4271e758ca..adc00a9098 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -56,8 +56,8 @@ use windmill_common::{ HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING, - OTEL_TRACING_PROXY_SETTING, NUGET_CONFIG_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING, - POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + NUGET_CONFIG_SETTING, OTEL_SETTING, OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING, + POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING, }, @@ -84,10 +84,10 @@ use windmill_common::{ use windmill_common::{client::AuthedClient, global_settings::APP_WORKSPACED_ROUTE_SETTING}; use windmill_queue::{cancel_job, get_queued_job_v2, SameWorkerPayload}; use windmill_worker::{ - handle_job_error, JobCompletedSender, OtelTracingProxySettings, SameWorkerSender, BUNFIG_INSTALL_SCOPES, - INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS, - OTEL_TRACING_PROXY_SETTINGS, NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, PIP_EXTRA_INDEX_URL, - PIP_INDEX_URL, POWERSHELL_REPO_PAT, POWERSHELL_REPO_URL, + handle_job_error, JobCompletedSender, OtelTracingProxySettings, SameWorkerSender, + BUNFIG_INSTALL_SCOPES, INSTANCE_PYTHON_VERSION, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR, MAVEN_REPOS, + NO_DEFAULT_MAVEN, NPM_CONFIG_REGISTRY, NUGET_CONFIG, OTEL_TRACING_PROXY_SETTINGS, + PIP_EXTRA_INDEX_URL, PIP_INDEX_URL, POWERSHELL_REPO_PAT, POWERSHELL_REPO_URL, }; #[cfg(feature = "parquet")] @@ -246,6 +246,7 @@ pub async fn initial_load( ), priority_tags_sorted: vec![], dedicated_worker: None, + dedicated_workers: None, init_bash: load_init_bash_from_env(), periodic_script_bash: load_periodic_bash_script_from_env(), periodic_script_interval_seconds: load_periodic_bash_script_interval_from_env(), @@ -784,26 +785,24 @@ pub async fn load_keep_job_dir(conn: &Connection) { pub async fn reload_otel_tracing_proxy_setting(conn: &Connection) { match load_value_from_global_settings_with_conn(conn, OTEL_TRACING_PROXY_SETTING, true).await { - Ok(Some(settings)) => { - match serde_json::from_value::(settings) { - Ok(new_settings) => { - let mut current = OTEL_TRACING_PROXY_SETTINGS.write().await; - if current.enabled != new_settings.enabled - || current.enabled_languages != new_settings.enabled_languages - { - tracing::info!( - "OTEL tracing proxy settings changed: enabled={}, languages={:?}", - new_settings.enabled, - new_settings.enabled_languages - ); - *current = new_settings; - } - } - Err(e) => { - tracing::error!("Error parsing OTEL tracing proxy settings: {e:#}"); + Ok(Some(settings)) => match serde_json::from_value::(settings) { + Ok(new_settings) => { + let mut current = OTEL_TRACING_PROXY_SETTINGS.write().await; + if current.enabled != new_settings.enabled + || current.enabled_languages != new_settings.enabled_languages + { + tracing::info!( + "OTEL tracing proxy settings changed: enabled={}, languages={:?}", + new_settings.enabled, + new_settings.enabled_languages + ); + *current = new_settings; } } - } + Err(e) => { + tracing::error!("Error parsing OTEL tracing proxy settings: {e:#}"); + } + }, Err(e) => { tracing::error!("Error loading OTEL tracing proxy setting: {e:#}"); } @@ -985,7 +984,10 @@ pub async fn delete_expired_items(db: &DB) -> () { ); } } - Err(e) => tracing::error!("Error deleting expired MCP OAuth authorization codes: {:?}", e), + Err(e) => tracing::error!( + "Error deleting expired MCP OAuth authorization codes: {:?}", + e + ), } let job_retention_secs = *JOB_RETENTION_SECS.read().await; @@ -1866,7 +1868,13 @@ pub async fn monitor_db( let update_min_worker_version_f = async { #[cfg(not(feature = "test_job_debouncing"))] - windmill_common::min_version::update_min_version(conn, _worker_mode, WORKERS_NAMES.read().await.clone(), initial_load).await; + windmill_common::min_version::update_min_version( + conn, + _worker_mode, + WORKERS_NAMES.read().await.clone(), + initial_load, + ) + .await; }; // Run every 5 minutes (10 iterations * 30s = 5 minutes) @@ -2060,10 +2068,12 @@ pub async fn reload_worker_config(db: &DB, tx: KillpillSender, kill_if_change: b } else { let wc = WORKER_CONFIG.read().await; let config = config.unwrap(); - if *wc != config || config.dedicated_worker.is_some() { + let has_dedicated = config.dedicated_worker.is_some() || config.dedicated_workers.as_ref().is_some_and(|dws| !dws.is_empty()); + if *wc != config || has_dedicated { if kill_if_change { - if config.dedicated_worker.is_some() + if has_dedicated || (*wc).dedicated_worker != config.dedicated_worker + || (*wc).dedicated_workers != config.dedicated_workers { tracing::info!("Dedicated worker config changed, sending killpill. Expecting to be restarted by supervisor."); let _ = tx.send(); diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 864acd0236..5ec7fa7263 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -995,6 +995,7 @@ TABLE: worker_ping - custom_tags (text[]) - worker_group (character) - dedicated_worker (character) + - dedicated_workers (text[]) - wm_version (character) - current_job_id (uuid) - current_job_workspace_id (character) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 8cac139f5d..f36cf81367 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5751,6 +5751,14 @@ paths: If true, the description field will be omitted from the response. schema: type: boolean + - name: dedicated_worker + in: query + description: | + (default regardless) + If true, show only scripts with dedicated_worker enabled. + If false, show only scripts with dedicated_worker disabled. + schema: + type: boolean responses: "200": @@ -7178,6 +7186,14 @@ paths: If true, the description field will be omitted from the response. schema: type: boolean + - name: dedicated_worker + in: query + description: | + (default regardless) + If true, show only flows with dedicated_worker enabled. + If false, show only flows with dedicated_worker disabled. + schema: + type: boolean responses: "200": description: All flow diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 83a2f9332a..e4cf392bf4 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -191,6 +191,9 @@ async fn list_flows( if !lq.include_draft_only.unwrap_or(false) || authed.is_operator { sqlb.and_where("o.draft_only IS NOT TRUE"); } + if let Some(dw) = &lq.dedicated_worker { + sqlb.and_where_eq("dedicated_worker", dw); + } if lq.with_deployment_msg.unwrap_or(false) { sqlb.join("deployment_metadata dm") diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 37698258bd..0f88589a97 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -371,6 +371,9 @@ async fn list_scripts( if let Some(it) = &lq.is_template { sqlb.and_where_eq("is_template", it); } + if let Some(dw) = &lq.dedicated_worker { + sqlb.and_where_eq("dedicated_worker", dw); + } if authed.is_operator { sqlb.and_where_eq("kind", quote("script")); } else if let Some(lowercased_kinds) = lowercased_kinds { diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index bb03e0db7a..420530ecf3 100644 --- a/backend/windmill-common/src/flows.rs +++ b/backend/windmill-common/src/flows.rs @@ -1146,6 +1146,7 @@ pub struct ListFlowQuery { pub starred_only: Option, pub include_draft_only: Option, pub with_deployment_msg: Option, + pub dedicated_worker: Option, } pub fn add_virtual_items_if_necessary(modules: &mut Vec) { diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index a7b8fb681a..28572894d2 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -687,6 +687,7 @@ pub struct ListScriptQuery { pub with_deployment_msg: Option, #[serde(default, deserialize_with = "from_seq")] pub languages: Option>, + pub dedicated_worker: Option, } fn from_seq<'de, D>(deserializer: D) -> Result>, D::Error> diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 73e6c7b8c7..7934890f25 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -219,6 +219,7 @@ lazy_static::lazy_static! { worker_tags: Default::default(), priority_tags_sorted: Default::default(), dedicated_worker: Default::default(), + dedicated_workers: Default::default(), cache_clear: Default::default(), init_bash: Default::default(), periodic_script_bash: Default::default(), @@ -1243,6 +1244,7 @@ pub struct Ping { pub ip: Option, pub tags: Option>, pub dw: Option, + pub dws: Option>, pub version: Option, pub vcpus: Option, pub memory: Option, @@ -1298,6 +1300,7 @@ pub async fn update_ping_http( &insert_ping.ip.unwrap(), insert_ping.tags.unwrap_or_default().as_slice(), insert_ping.dw, + insert_ping.dws.as_deref(), &insert_ping.version.unwrap(), insert_ping.vcpus, insert_ping.memory, @@ -1428,6 +1431,7 @@ pub async fn insert_ping_query( ip: &str, tags: &[String], dw: Option, + dws: Option<&[String]>, version: &str, vcpus: Option, memory: Option, @@ -1435,14 +1439,15 @@ pub async fn insert_ping_query( db: &DB, ) -> anyhow::Result<()> { sqlx::query!( - "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON CONFLICT (worker) - DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group", + "INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, dedicated_workers, wm_version, vcpus, memory, job_isolation) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT (worker) + DO UPDATE set ip = EXCLUDED.ip, custom_tags = EXCLUDED.custom_tags, worker_group = EXCLUDED.worker_group, dedicated_workers = EXCLUDED.dedicated_workers", worker_instance, worker_name, ip, tags, worker_group, dw, + dws, version, vcpus, memory, @@ -1614,6 +1619,30 @@ pub async fn load_worker_config( } }) .transpose()?; + + // Parse dedicated_workers (multiple dedicated workers) + let dedicated_workers = config + .dedicated_workers + .map(|workers| { + workers + .into_iter() + .map(|x| { + let splitted = x.split(':').to_owned().collect_vec(); + if splitted.len() != 2 { + return Err(anyhow::anyhow!( + "Invalid dedicated_workers format. Got {x}, expects :" + )); + } + let workspace = splitted[0]; + let script_path = splitted[1]; + Ok(WorkspacedPath { + workspace_id: workspace.to_string(), + path: script_path.to_string(), + }) + }) + .collect::, _>>() + }) + .transpose()?; if *WORKER_GROUP == "default" && dedicated_worker.is_none() { let mut all_tags = config .worker_tags @@ -1647,7 +1676,18 @@ pub async fn load_worker_config( let worker_tags = config .worker_tags .or_else(|| { - if let Some(ref dedicated_worker) = dedicated_worker.as_ref() { + // Check for multiple dedicated workers first + if let Some(ref dws) = dedicated_workers.as_ref() { + let mut dedi_tags: Vec = dws + .iter() + .map(|dw| format!("{}:{}", dw.workspace_id, dw.path)) + .collect(); + if std::env::var("ADD_FLOW_TAG").is_ok() { + dedi_tags.push("flow".to_string()); + } + Some(dedi_tags) + } else if let Some(ref dedicated_worker) = dedicated_worker.as_ref() { + // Fallback to single dedicated worker for backward compatibility let mut dedi_tags = vec![format!( "{}:{}", dedicated_worker.workspace_id, dedicated_worker.path @@ -1743,6 +1783,7 @@ pub async fn load_worker_config( worker_tags, priority_tags_sorted, dedicated_worker, + dedicated_workers, init_bash: config .init_bash .or_else(|| load_init_bash_from_env()) @@ -1836,6 +1877,7 @@ pub struct WorkerConfigOpt { pub worker_tags: Option>, pub priority_tags: Option>, pub dedicated_worker: Option, + pub dedicated_workers: Option>, pub init_bash: Option, pub periodic_script_bash: Option, pub periodic_script_interval_seconds: Option, @@ -1852,6 +1894,7 @@ impl Default for WorkerConfigOpt { worker_tags: Default::default(), priority_tags: Default::default(), dedicated_worker: Default::default(), + dedicated_workers: Default::default(), init_bash: Default::default(), periodic_script_bash: Default::default(), periodic_script_interval_seconds: Default::default(), @@ -1869,6 +1912,7 @@ pub struct WorkerConfig { pub worker_tags: Vec, pub priority_tags_sorted: Vec, pub dedicated_worker: Option, + pub dedicated_workers: Option>, pub init_bash: Option, pub periodic_script_bash: Option, pub periodic_script_interval_seconds: Option, @@ -1880,8 +1924,8 @@ pub struct WorkerConfig { 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: {:?}, init_bash: {:?}, periodic_script_bash: {:?}, periodic_script_interval_seconds: {:?}, cache_clear: {:?}, additional_python_paths: {:?}, pip_local_dependencies: {:?}, env_vars: {:?} }}", - self.worker_tags, self.priority_tags_sorted, self.dedicated_worker, 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::>().join(", ")) + 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: {:?} }}", + 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::>().join(", ")) } } diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 79dd74bf25..2d2d143eba 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -533,6 +533,7 @@ pub async fn update_worker_ping_for_failed_init_script( ip: None, tags: None, dw: None, + dws: None, jobs_executed: None, occupancy_rate: None, occupancy_rate_15s: None, diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e9a4ebc145..c522630354 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -1529,7 +1529,10 @@ pub async fn run_worker( let mut occupancy_metrics = OccupancyMetrics::new(start_time); let mut jobs_executed = 0; - let is_dedicated_worker: bool = WORKER_CONFIG.read().await.dedicated_worker.is_some(); + let is_dedicated_worker: bool = { + let config = WORKER_CONFIG.read().await; + config.dedicated_worker.is_some() || config.dedicated_workers.as_ref().is_some_and(|dws| !dws.is_empty()) + }; #[cfg(feature = "benchmark")] let benchmark_jobs: i32 = std::env::var("BENCHMARK_JOBS") @@ -1631,9 +1634,9 @@ pub async fn run_worker( // Option>, #[cfg(all(feature = "private", feature = "enterprise"))] - let (dedicated_workers, is_flow_worker, dedicated_handles): ( + let (dedicated_workers, dedicated_flow_paths, dedicated_handles): ( HashMap>, - bool, + HashSet, Vec>, ) = match conn { Connection::Sql(pool) => { @@ -1648,15 +1651,15 @@ pub async fn run_worker( ) .await } - Connection::Http(_) => (HashMap::new(), false, vec![]), + Connection::Http(_) => (HashMap::new(), HashSet::new(), vec![]), }; #[cfg(any(not(feature = "private"), not(feature = "enterprise")))] - let (dedicated_workers, is_flow_worker, dedicated_handles): ( + let (dedicated_workers, dedicated_flow_paths, dedicated_handles): ( HashMap>, - bool, + HashSet, Vec>, - ) = (HashMap::new(), false, vec![]); + ) = (HashMap::new(), HashSet::new(), vec![]); if i_worker == 1 { if let Err(e) = queue_init_bash_maybe(conn, same_worker_tx.clone(), &worker_name).await { @@ -2030,30 +2033,32 @@ pub async fn run_worker( JobKind::Script | JobKind::Preview | JobKind::FlowScript ) { if !dedicated_workers.is_empty() { - let key_o = if is_flow_worker { - job.flow_step_id.as_ref().map(|x| x.to_string()) + // Try flow path + step_id combinations for flow jobs, otherwise use runnable_path + let dedicated_worker_tx = if let Some(step_id) = job.flow_step_id.as_ref() { + dedicated_flow_paths.iter().find_map(|flow_path| { + let key = format!("{}:{}", flow_path, step_id); + dedicated_workers.get(&key) + }) } else { - job.runnable_path.as_ref().map(|x| x.to_string()) + job.runnable_path.as_ref().and_then(|path| dedicated_workers.get(path)) }; - if let Some(key) = key_o { - if let Some(dedicated_worker_tx) = dedicated_workers.get(&key) { - let dedicated_job = DedicatedWorkerJob { - job: Arc::new(job.job()), - flow_runners: None, - done_tx: None, - }; - if let Err(e) = dedicated_worker_tx.send(dedicated_job).await { - tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}"); - } - - #[cfg(feature = "benchmark")] - { - add_time!(bench, "sent to dedicated worker"); - infos.add_iter(bench, true); - } - - continue; + if let Some(dedicated_worker_tx) = dedicated_worker_tx { + let dedicated_job = DedicatedWorkerJob { + job: Arc::new(job.job()), + flow_runners: None, + done_tx: None, + }; + if let Err(e) = dedicated_worker_tx.send(dedicated_job).await { + tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}"); } + + #[cfg(feature = "benchmark")] + { + add_time!(bench, "sent to dedicated worker"); + infos.add_iter(bench, true); + } + + continue; } } diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index e92092480b..ffb4f3054e 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/backend/windmill-worker/src/worker_utils.rs @@ -133,6 +133,7 @@ async fn update_worker_ping_full_inner( ip: None, tags: Some(tags.to_vec()), dw: None, + dws: None, jobs_executed: Some(jobs_executed), occupancy_rate: Some(occupancy_rate), occupancy_rate_15s: Some(occupancy_rate_15s.unwrap_or(0.0)), @@ -159,13 +160,19 @@ pub async fn insert_ping( ip: &str, db: &Connection, ) -> anyhow::Result<()> { - let (tags, dw) = { + let (tags, dw, dws) = { let wc = WORKER_CONFIG.read().await.clone(); ( wc.worker_tags, wc.dedicated_worker .as_ref() .map(|x| format!("{}:{}", x.workspace_id, x.path)), + wc.dedicated_workers.as_ref().map(|workers| { + workers + .iter() + .map(|x| format!("{}:{}", x.workspace_id, x.path)) + .collect::>() + }), ) }; @@ -190,6 +197,7 @@ pub async fn insert_ping( ip, tags.as_slice(), dw, + dws.as_deref(), windmill_common::utils::GIT_VERSION, vcpus, memory, @@ -210,6 +218,7 @@ pub async fn insert_ping( ip: Some(ip.to_string()), tags: Some(tags.to_vec()), dw: dw, + dws: dws, jobs_executed: None, occupancy_rate: None, occupancy_rate_15s: None, @@ -282,6 +291,7 @@ pub async fn update_worker_ping_from_job( ip: None, tags: None, dw: None, + dws: None, version: None, vcpus: None, memory: None, diff --git a/frontend/src/lib/components/DedicatedWorkersSelector.svelte b/frontend/src/lib/components/DedicatedWorkersSelector.svelte new file mode 100644 index 0000000000..df37b5bf17 --- /dev/null +++ b/frontend/src/lib/components/DedicatedWorkersSelector.svelte @@ -0,0 +1,645 @@ + + +
+ + {#if selectedTags.length > 0} +
+
+ {#each selectedTags as tag (tag)} + {@const info = selectedTagsInfo.get(tag)} +
+
+ {#if info?.type === 'flow' && info.runners && info.runners.length > 0} + + {:else} +
+ {/if} +
+ {#if info} + {#if info.type === 'flow'} + + {:else} + + {/if} + {info.path} + ({info.workspace}) + {#if info.type === 'flow' && info.runners} + + {info.runners.length} runner{info.runners.length !== 1 ? 's' : ''} + + {:else if info.type === 'script'} + 1 runner + {/if} + {:else} + {tag} + {/if} +
+ {#if !disabled} + + {/if} +
+ + {#if info?.type === 'flow' && info.expanded && info.runners} +
+ {#each info.runners as runner (runner.stepId)} +
+ {runner.stepId} + {#if runner.stepSummary} + {runner.stepSummary} + {/if} + + {runner.isInline ? runner.language : runner.scriptPath} + +
+ {/each} +
+ {/if} +
+ {/each} +
+
+ {/if} + + +
+ + + {#if selectorExpanded} +
+ +
+ Workspace + {}} - bind:value={nconfig.dedicated_worker} - />
+ +

Workers will get killed upon detecting changes. It is assumed they are in an environment + where the supervisor will restart them.

+ + {#if nconfig !== undefined} + { + if (nconfig) { + nconfig.dedicated_workers = tags + nconfig.dedicated_worker = undefined + } + }} + /> {/if}
{/if}