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

fix: improve dedicated workers

Browse Source
This commit is contained in:
Ruben Fiszel
2023-11-06 11:21:36 +01:00
parent 34da785a60
commit aeacaa2b01
11 changed files with 121 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/worker.rs
+11 -5
View File
@@ -1658,7 +1658,8 @@ func main(derp string) (string, error) {
language: ScriptLang::Go,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
cache_ttl: None,
dedicated_worker: None
}))
.arg("derp", json!("world"))
.run_until_complete(&db, port)
@@ -1688,7 +1689,8 @@ echo "hello $msg"
language: ScriptLang::Bash,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
cache_ttl: None,
dedicated_worker: None
}))
.arg("msg", json!("world"))
.run_until_complete(&db, port)
@@ -1715,7 +1717,8 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
cache_ttl: None,
dedicated_worker: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
@@ -1748,7 +1751,8 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
cache_ttl: None,
dedicated_worker: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
@@ -1780,7 +1784,8 @@ def main():
lock: None,
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None
cache_ttl: None,
dedicated_worker: None
});
let result = run_job_in_new_worker_until_complete(&db, job, port)
@@ -3177,6 +3182,7 @@ async fn run_preview_relative_imports(db: &Pool<Postgres>, script_content: Strin
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None
})).push(&db2).await;
backend/windmill-api/openapi.yaml
+2 -1
View File
@@ -7452,7 +7452,8 @@ components:
kind:
type: string
enum: [code, identity, http]
dedicated_worker:
type: boolean
required:
- args
backend/windmill-api/src/jobs.rs
+2
View File
@@ -1468,6 +1468,7 @@ struct Preview {
args: Option<Box<JsonRawValue>>,
language: Option<ScriptLang>,
tag: Option<String>,
dedicated_worker: Option<bool>,
}
#[derive(Deserialize)]
@@ -2262,6 +2263,7 @@ async fn run_preview_job(
concurrent_limit: None, // TODO(gbouv): once I find out how to store limits in the content of a script, should be easy to plug limits here
concurrency_time_window_s: None, // TODO(gbouv): same as above
cache_ttl: None,
dedicated_worker: preview.dedicated_worker,
}),
},
preview.args.unwrap_or_default(),
backend/windmill-common/src/jobs.rs
+1
View File
@@ -315,6 +315,7 @@ pub struct RawCode {
pub concurrent_limit: Option<i32>,
pub concurrency_time_window_s: Option<i32>,
pub cache_ttl: Option<i32>,
pub dedicated_worker: Option<bool>,
}
type Tag = String;
backend/windmill-common/src/worker.rs
+7 -2
View File
@@ -209,10 +209,14 @@ pub async fn load_worker_config(db: &DB) -> error::Result<WorkerConfig> {
.worker_tags
.or_else(|| {
if let Some(ref dedicated_worker) = dedicated_worker.as_ref() {
Some(vec![format!(
let mut dedi_tags = vec![format!(
"{}:{}",
dedicated_worker.workspace_id, dedicated_worker.path
)])
)];
if std::env::var("ADD_FLOW_TAG").is_ok() {
dedi_tags.push("flow".to_string());
}
Some(dedi_tags)
} else {
std::env::var("WORKER_TAGS")
.ok()
@@ -220,6 +224,7 @@ pub async fn load_worker_config(db: &DB) -> error::Result<WorkerConfig> {
}
})
.unwrap_or_else(|| DEFAULT_TAGS.clone());
let mut priority_tags_sorted: Vec<PriorityTags> = Vec::new();
let priority_tags_map = config.priority_tags.unwrap_or_else(HashMap::new);
if priority_tags_map.len() > 0 {
backend/windmill-queue/src/jobs.rs
+3 -2
View File
@@ -1453,7 +1453,7 @@ async fn pull_single_job_and_mark_as_running_no_concurrency_limit<
LIMIT 1
)
RETURNING *")
.bind(tags.clone())
.bind(tags)
.fetch_optional(db)
.await?
} else {
@@ -2208,6 +2208,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
dedicated_worker,
}) => (
None,
path,
@@ -2219,7 +2220,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
concurrent_limit,
concurrency_time_window_s,
cache_ttl,
None,
dedicated_worker,
None,
),
JobPayload::Dependencies { hash, dependencies, language, path } => (
backend/windmill-worker/src/bun_executor.rs
+27 -18
View File
@@ -577,28 +577,37 @@ pub async fn start_worker(
let _ = write_file(job_dir, "package.json", &splitted[0]).await?;
let lockb = splitted[1];
if lockb != EMPTY_FILE {
let _ = write_file_binary(
let has_trusted_deps = &splitted[0].contains("trustedDependencies");
if !has_trusted_deps {
let _ = write_file_binary(
job_dir,
"bun.lockb",
&base64::engine::general_purpose::STANDARD
.decode(&splitted[1])
.map_err(|_| {
error::Error::InternalErr("Could not decode bun.lockb".to_string())
})?,
)
.await?;
}
install_lockfile(
&mut logs,
&mut mem_peak,
&Uuid::nil(),
&w_id,
db,
job_dir,
"bun.lockb",
&base64::engine::general_purpose::STANDARD
.decode(&splitted[1])
.map_err(|_| {
error::Error::InternalErr("Could not decode bun.lockb".to_string())
})?,
worker_name,
common_bun_proc_envs.clone(),
)
.await?;
if !has_trusted_deps {
remove_dir_all(format!("{}/node_modules", job_dir)).await?;
}
tracing::info!("dedicated worker requirements installed: {reqs}");
}
install_lockfile(
&mut logs,
&mut mem_peak,
&Uuid::nil(),
&w_id,
db,
job_dir,
worker_name,
common_bun_proc_envs.clone(),
)
.await?;
} else if !*DISABLE_NSJAIL {
let trusted_deps = get_trusted_deps(inner_content);
logs.push_str("\n\n--- BUN INSTALL ---\n");
backend/windmill-worker/src/dedicated_worker.rs
+24 -4
View File
@@ -8,7 +8,7 @@ use tokio::{
io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
process::Command,
};
use windmill_common::{error, jobs::QueuedJob, variables};
use windmill_common::{error, jobs::QueuedJob, variables, worker::to_raw_value};
use std::{collections::VecDeque, process::Stdio, sync::Arc};
@@ -82,8 +82,15 @@ pub async fn handle_dedicated_process(
.take()
.expect("child did not have a handle to stdout");
let stderr = child
.stderr
.take()
.expect("child did not have a handle to stderr");
let mut reader = BufReader::new(stdout).lines();
let mut err_reader = BufReader::new(stderr).lines();
let mut stdin = child
.stdin
.take()
@@ -115,6 +122,14 @@ pub async fn handle_dedicated_process(
tracing::info!("Could not write end message to stdin: {e:?}")
}
},
line = err_reader.next_line() => {
if let Some(line) = line.expect("line is ok") {
tracing::error!("dedicated worker process stderr: {:?}", line);
} else {
tracing::info!("dedicated worker process exited");
break;
}
},
line = reader.next_line() => {
// j += 1;
@@ -123,11 +138,16 @@ pub async fn handle_dedicated_process(
tracing::info!("dedicated worker process started");
continue;
}
tracing::debug!("processed job");
tracing::debug!("processed job: {line}");
let result = serde_json::from_str(&line).expect("json is ok");
let job: Arc<QueuedJob> = jobs.pop_front().expect("pop");
job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), mem_peak: 0, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap();
match serde_json::from_str::<Box<serde_json::value::RawValue>>(&line) {
Ok(result) => job_completed_tx.send(JobCompleted { job , result, logs: "".to_string(), mem_peak: 0, success: true, cached_res_path: None, token: token.to_string() }).await.unwrap(),
Err(e) => {
tracing::error!("Could not deserialize job result `{line}`: {e:?}");
job_completed_tx.send(JobCompleted { job , result: to_raw_value(&serde_json::json!({"error": format!("Could not deserialize job result `{line}`: {e:?}")})), logs: "".to_string(), mem_peak: 0, success: false, cached_res_path: None, token: token.to_string() }).await.unwrap();
},
};
} else {
tracing::info!("dedicated worker process exited");
break;
backend/windmill-worker/src/worker.rs
+39 -23
View File
@@ -35,7 +35,8 @@ use windmill_common::{
users::SUPERADMIN_SECRET_EMAIL,
utils::{rd_string, StripPath},
worker::{
to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, WORKER_CONFIG, WORKER_GROUP,
to_raw_value, to_raw_value_owned, update_ping, WorkspacedPath, CLOUD_HOSTED, WORKER_CONFIG,
WORKER_GROUP,
},
DB, IS_READY, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
};
@@ -1079,7 +1080,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
IS_READY.store(true, Ordering::Relaxed);
tracing::info!(worker = %worker_name, "listening for jobs, WORKER_GROUP: {}, config: {:?}", *WORKER_GROUP, WORKER_CONFIG.read().await);
let (dedicated_worker_tx, dedicated_worker_handle) = if let Some(_wp) =
let (dedi_path, dedicated_worker_tx, dedicated_worker_handle) = if let Some(_wp) =
WORKER_CONFIG.read().await.dedicated_worker.clone()
{
#[cfg(not(feature = "enterprise"))]
@@ -1089,6 +1090,8 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
return;
}
let dedi_path = _wp.clone();
#[cfg(feature = "enterprise")]
{
let (dedicated_worker_tx, dedicated_worker_rx) =
@@ -1208,10 +1211,15 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tracing::error!("error in dedicated worker: {:?}", e)
}
});
(Some(dedicated_worker_tx), Some(handle))
(Some(dedi_path), Some(dedicated_worker_tx), Some(handle))
}
} else {
(None, None) as (Option<Sender<Arc<QueuedJob>>>, Option<JoinHandle<()>>)
(None, None, None)
as (
Option<WorkspacedPath>,
Option<Sender<Arc<QueuedJob>>>,
Option<JoinHandle<()>>,
)
};
#[cfg(feature = "benchmark")]
@@ -1447,28 +1455,35 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
last_executed_job = None;
jobs_executed += 1;
if let Some(dedicated_worker_tx) = dedicated_worker_tx.clone() {
#[cfg(feature = "benchmark")]
main_duration
.fetch_add(loop_start.elapsed().as_millis() as usize, Ordering::SeqCst);
#[cfg(feature = "benchmark")]
let send_start = Instant::now();
if let (Some(dedi_path), Some(dedicated_worker_tx)) =
(dedi_path.as_ref(), dedicated_worker_tx.clone())
{
if dedi_path.workspace_id == job.workspace_id
&& Some(&dedi_path.path) == job.script_path.as_ref()
{
#[cfg(feature = "benchmark")]
main_duration
.fetch_add(loop_start.elapsed().as_millis() as usize, Ordering::SeqCst);
#[cfg(feature = "benchmark")]
let send_start = Instant::now();
let timer = worker_dedicated_channel_queue_send_duration
.as_ref()
.map(|x| x.start_timer());
let timer = worker_dedicated_channel_queue_send_duration
.as_ref()
.map(|x| x.start_timer());
if let Err(e) = dedicated_worker_tx.send(Arc::new(job)).await {
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
if let Err(e) = dedicated_worker_tx.send(Arc::new(job)).await {
tracing::info!("failed to send jobs to dedicated workers. Likely dedicated worker has been shut down. This is normal: {e:?}");
}
timer.map(|x| x.stop_and_record());
#[cfg(feature = "benchmark")]
send_duration
.fetch_add(send_start.elapsed().as_millis() as usize, Ordering::SeqCst);
continue;
}
timer.map(|x| x.stop_and_record());
#[cfg(feature = "benchmark")]
send_duration
.fetch_add(send_start.elapsed().as_millis() as usize, Ordering::SeqCst);
continue;
} else if matches!(job.job_kind, JobKind::Noop) {
}
if matches!(job.job_kind, JobKind::Noop) {
#[cfg(feature = "benchmark")]
main_duration
.fetch_add(loop_start.elapsed().as_millis() as usize, Ordering::SeqCst);
@@ -1726,6 +1741,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
concurrent_limit: None,
concurrency_time_window_s: None,
cache_ttl: None,
dedicated_worker: None,
}),
PushArgs::empty(),
worker_name,
backend/windmill-worker/src/worker_flow.rs
+4 -1
View File
@@ -1426,7 +1426,9 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
match json_value {
Ok(serde_json::Value::Number(n)) => {
if !n.is_u64() {
return Err(Error::ExecutionErr(format!("Expected an integer, found: {n}")));
return Err(Error::ExecutionErr(format!(
"Expected an integer, found: {n}"
)));
}
n.as_u64().map(|x| from_now(Duration::from_secs(x)))
@@ -2554,6 +2556,7 @@ fn raw_script_to_payload(
concurrent_limit: *concurrent_limit,
concurrency_time_window_s: *concurrency_time_window_s,
cache_ttl: module.cache_ttl.map(|x| x as i32),
dedicated_worker: None,
}),
tag: tag.clone(),
}
frontend/src/lib/components/sidebar/UserMenu.svelte
+1 -1
View File
@@ -1,7 +1,7 @@
<script lang="ts">
import { goto } from '$app/navigation'
import { logout } from '$lib/logout'
import { userStore, usersWorkspaceStore, superadmin, usageStore, premiumStore } from '$lib/stores'
import { userStore, usersWorkspaceStore, usageStore, premiumStore } from '$lib/stores'
import { faCog, faCrown, faHardHat, faSignOut } from '@fortawesome/free-solid-svg-icons'
import Icon from 'svelte-awesome'
import Menu from '../common/menu/MenuV2.svelte'