diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 7b89d6aaa5..db755cfb53 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -40,4 +40,4 @@ jobs: backend -> target - name: cargo test timeout-minutes: 10 - run: mkdir frontend/build && cd backend && touch windmill-api/openapi-deref.yaml && DATABASE_URL=postgres://postgres:changeme@postgres:5432/windmill cargo test --all -- --nocapture + run: mkdir frontend/build && cd backend && touch windmill-api/openapi-deref.yaml && DATABASE_URL=postgres://postgres:changeme@postgres:5432/windmill DISABLE_NSJAIL=false cargo test --all -- --nocapture diff --git a/README.md b/README.md index 2fdcbe4014..70185aca2e 100644 --- a/README.md +++ b/README.md @@ -302,7 +302,8 @@ you to have it being synced automatically everyday. | Environment Variable name | Default | Description | Api Server/Worker/All | | ------------------------- | ---------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------- | | DATABASE_URL | | The Postgres database url. | All | -| DISABLE_NSJAIL | true | Disable Nsjail Sandboxing | | +| DISABLE_NSJAIL | true | Disable Nsjail Sandboxing | Worker | +| PORT | 8000 | Exposed port | Server | | | NUM_WORKERS | 3 | The number of worker per Worker instance (set to 1 on Eks to have 1 pod = 1 worker, set to 0 for an API only instance) | Worker | | DISABLE_SERVER | false | Binary would operate as a worker only instance | Worker | | METRICS_ADDR | None | The socket addr at which to expose Prometheus metrics at the /metrics path. Set to "true" to expose it on port 8001 | All | @@ -318,8 +319,7 @@ you to have it being synced automatically everyday. | S3_CACHE_BUCKET (EE only) | None | The S3 bucket to sync the cache of the workers to | Worker | | TAR_CACHE_RATE (EE only) | 100 | The rate at which to tar the cache of the workers. 100 means every 100th job in average (uniformly randomly distributed). | Worker | | SLACK_SIGNING_SECRET | None | The signing secret of your Slack app. See [Slack documentation](https://api.slack.com/authentication/verifying-requests-from-slack) | Server | -| COOKIE_DOMAIN | None | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin | Server | -| SERVE_CSP | None | The CSP directives to use when serving the frontend static assets | Server | +| COOKIE_DOMAIN | None | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin | Server | | | DENO_PATH | /usr/bin/deno | The path to the deno binary. | Worker | | PYTHON_PATH | /usr/local/bin/python3 | The path to the python binary. | Worker | | GO_PATH | /usr/bin/go | The path to the go binary. | Worker | diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 94d18ee9e0..cb12335780 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -4785,6 +4785,7 @@ dependencies = [ "hex", "hmac", "hyper", + "lazy_static", "prometheus", "rand 0.8.5", "reqwest", diff --git a/backend/src/main.rs b/backend/src/main.rs index faf87424e4..99257a2b37 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -11,9 +11,10 @@ use std::net::SocketAddr; use git_version::git_version; use sqlx::{Pool, Postgres}; use windmill_common::utils::rd_string; -use windmill_worker::WorkerConfig; const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); +const DEFAULT_NUM_WORKERS: usize = 3; +const DEFAULT_PORT: u16 = 8000; mod ee; @@ -26,7 +27,7 @@ async fn main() -> anyhow::Result<()> { let num_workers = std::env::var("NUM_WORKERS") .ok() .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_NUM_WORKERS as i32); + .unwrap_or(DEFAULT_NUM_WORKERS as i32); let metrics_addr: Option = std::env::var("METRICS_ADDR") .ok() @@ -38,6 +39,13 @@ async fn main() -> anyhow::Result<()> { .transpose()? .flatten(); + let port: u16 = std::env::var("PORT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(DEFAULT_PORT as u16); + let base_internal_url: String = std::env::var("BASE_INTERNAL_URL") + .unwrap_or_else(|_| format!("http://localhost:{}", port.to_string())); + let server_mode = !std::env::var("DISABLE_SERVER") .ok() .and_then(|x| x.parse::().ok()) @@ -52,58 +60,25 @@ async fn main() -> anyhow::Result<()> { let (tx, rx) = tokio::sync::broadcast::channel::<()>(3); let shutdown_signal = windmill_common::shutdown_signal(tx); - let base_url = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); - - let base_internal_url = - std::env::var("BASE_INTERNAL_URL").unwrap_or_else(|_| "http://localhost:8000".to_string()); - let timeout = std::env::var("TIMEOUT") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_TIMEOUT); - if server_mode || num_workers > 0 { - let addr = SocketAddr::from(([0, 0, 0, 0], 8000)); + let addr = SocketAddr::from(([0, 0, 0, 0], port)); - let base_url2 = base_url.clone(); let server_f = async { if server_mode { - windmill_api::run_server(db.clone(), addr, base_url, rx.resubscribe()).await?; + windmill_api::run_server(db.clone(), addr, rx.resubscribe()).await?; } Ok(()) as anyhow::Result<()> }; - let base_url = base_url2.clone(); let workers_f = async { if num_workers > 0 { - let sleep_queue = std::env::var("SLEEP_QUEUE") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_SLEEP_QUEUE); - let disable_nuser = std::env::var("DISABLE_NUSER") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false); - let disable_nsjail = std::env::var("DISABLE_NSJAIL") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(true); - let keep_job_dir = std::env::var("KEEP_JOB_DIR") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false); - let license_key = std::env::var("LICENSE_KEY").ok(); - let sync_bucket = std::env::var("S3_CACHE_BUCKET") - .ok() - .map(|e| Some(e)) - .unwrap_or(None); - #[cfg(feature = "enterprise")] tracing::info!( - " + " ############################## -Windmill Enterprise Edition {GIT_VERSION} LICENSE_KEY: {license_key:?}, S3_CACHE_BUCKET: {sync_bucket:?} +Windmill Enterprise Edition {GIT_VERSION} ##############################" - ); + ); #[cfg(not(feature = "enterprise"))] tracing::info!( @@ -113,38 +88,58 @@ Windmill Community Edition {GIT_VERSION} ##############################" ); - tracing::info!( - "DISABLE_NSJAIL: {disable_nsjail}, DISABLE_NUSER: {disable_nuser}, BASE_URL: \ - {base_url}, SLEEP_QUEUE: {sleep_queue}, NUM_WORKERS: {num_workers}, TIMEOUT: \ - {timeout}, KEEP_JOB_DIR: {keep_job_dir}" - ); + display_config(vec![ + "DISABLE_NSJAIL", + "DISABLE_SERVER", + "NUM_WORKERS", + "METRICS_ADDR", + "JSON_FMT", + "BASE_URL", + "BASE_INTERNAL_URL", + "TIMEOUT", + "SLEEP_QUEUE", + "MAX_LOG_SIZE", + "PORT", + "KEEP_JOB_DIR", + "S3_CACHE_BUCKET", + "TAR_CACHE_RATE", + "COOKIE_DOMAIN", + "PYTHON_PATH", + "DENO_PATH", + "GO_PATH", + "PIP_INDEX_URL", + "PIP_EXTRA_INDEX_URL", + "PIP_TRUSTED_HOST", + "PATH", + "HOME", + "DATABASE_CONNECTIONS", + "TIMEOUT_WAIT_RESULT", + "QUEUE_LIMIT_WAIT_RESULT", + "DENO_AUTH_TOKENS", + "DENO_FLAGS", + "PIP_LOCAL_DEPENDENCIES", + "ADDITIONAL_PYTHON_PATHS", + "INCLUDE_HEADERS", + "WHITELIST_WORKSPACES", + "BLACKLIST_WORKSPACES", + "NEW_USER_WEBHOOK", + "CLOUD_HOSTED", + ]); run_workers( db.clone(), - addr, - timeout, - num_workers, - sleep_queue, - WorkerConfig { - disable_nsjail, - disable_nuser, - base_internal_url, - base_url, - keep_job_dir, - }, rx.resubscribe(), - sync_bucket, - license_key, + num_workers, + base_internal_url.clone(), ) .await?; } Ok(()) as anyhow::Result<()> }; - let base_url = base_url2; let monitor_f = async { if server_mode { - monitor_db(&db, timeout, base_url, rx.resubscribe()); + monitor_db(&db, rx.resubscribe(), &base_internal_url); } Ok(()) as anyhow::Result<()> }; @@ -163,34 +158,46 @@ Windmill Community Edition {GIT_VERSION} Ok(()) } +fn display_config(envs: Vec<&str>) { + tracing::info!( + "config: {}", + envs.iter() + .filter(|env| std::env::var(env).is_ok()) + .map(|env| { + format!( + "{}: {}", + env, + std::env::var(env).unwrap_or_else(|_| "not set".to_string()) + ) + }) + .collect::>() + .join(", ") + ) +} + pub fn monitor_db( db: &Pool, - timeout: i32, - base_url: String, rx: tokio::sync::broadcast::Receiver<()>, + base_internal_url: &str, ) { let db1 = db.clone(); let db2 = db.clone(); let rx2 = rx.resubscribe(); - + let base_internal_url = base_internal_url.to_string(); tokio::spawn(async move { - windmill_worker::handle_zombie_jobs_periodically(&db1, timeout, &base_url, rx).await + windmill_worker::handle_zombie_jobs_periodically(&db1, rx, &base_internal_url).await }); tokio::spawn(async move { windmill_api::delete_expired_items_perdiodically(&db2, rx2).await }); } pub async fn run_workers( db: Pool, - addr: SocketAddr, - timeout: i32, - num_workers: i32, - sleep_queue: u64, - worker_config: WorkerConfig, rx: tokio::sync::broadcast::Receiver<()>, - mut periodic_script: Option, - license_key: Option, + num_workers: i32, + base_internal_url: String, ) -> anyhow::Result<()> { + let license_key = std::env::var("LICENSE_KEY").ok(); #[cfg(feature = "enterprise")] ee::verify_license_key(license_key)?; @@ -198,12 +205,6 @@ pub async fn run_workers( if license_key.is_some() { panic!("License key is required ONLY for the enterprise edition"); } - #[cfg(not(feature = "enterprise"))] - if !worker_config.disable_nsjail { - tracing::warn!( - "NSJAIL to sandbox process in untrusted environments is an enterprise feature but allowed to be used for testing purposes" - ); - } let instance_name = rd_string(5); let monitor = tokio_metrics::TaskMonitor::new(); @@ -223,22 +224,17 @@ pub async fn run_workers( let worker_name = format!("dt-worker-{}-{}", &instance_name, rd_string(5)); let ip = ip.clone(); let rx = rx.resubscribe(); - let worker_config = worker_config.clone(); - let wp = periodic_script.take(); + let base_internal_url = base_internal_url.clone(); handles.push(tokio::spawn(monitor.instrument(async move { - tracing::info!(addr = %addr.to_string(), worker = %worker_name, "starting worker"); + tracing::info!(worker = %worker_name, "starting worker"); windmill_worker::run_worker( &db1, - timeout, &instance_name, worker_name, i as u64, - num_workers as u64, &ip, - sleep_queue, - worker_config, - wp, rx, + &base_internal_url, ) .await }))); diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 31f6a1d120..7a3b2d508f 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -6,10 +6,8 @@ use windmill_common::{ flow_status::{FlowStatus, FlowStatusModule}, flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform}, scripts::ScriptLang, - DEFAULT_SLEEP_QUEUE, }; use windmill_queue::{get_queued_job, JobPayload, RawCode}; -use windmill_worker::WorkerConfig; async fn initialize_tracing() { use std::sync::Once; @@ -89,14 +87,7 @@ impl ApiServer { let addr = sock.local_addr().unwrap(); drop(sock); - let task = tokio::task::spawn({ - windmill_api::run_server( - db.clone(), - addr, - format!("http://localhost:{}", addr.port()), - rx, - ) - }); + let task = tokio::task::spawn(windmill_api::run_server(db.clone(), addr, rx)); return Self { addr, tx, task }; } @@ -917,43 +908,20 @@ fn spawn_test_worker( ) { let (tx, rx) = tokio::sync::broadcast::channel(1); let db = db.to_owned(); - let timeout = 4_000; let worker_instance: &str = "test worker instance"; let worker_name: String = next_worker_name(); let i_worker: u64 = Default::default(); - let num_workers: u64 = 2; let ip: &str = Default::default(); - let sleep_queue: u64 = DEFAULT_SLEEP_QUEUE / num_workers; - let port = port; - let worker_config = WorkerConfig { - base_internal_url: format!("http://localhost:{port}"), - base_url: format!("http://localhost:{port}"), - disable_nuser: std::env::var("DISABLE_NUSER") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false), - disable_nsjail: std::env::var("DISABLE_NSJAIL") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false), - keep_job_dir: std::env::var("KEEP_JOB_DIR") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false), - }; let future = async move { + let base_internal_url = format!("http://localhost:{}", port); windmill_worker::run_worker( &db, - timeout, worker_instance, worker_name, i_worker, - num_workers, ip, - sleep_queue, - worker_config, - None, rx, + &base_internal_url, ) .await }; diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 936284f177..80a497d607 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -8,10 +8,6 @@ edition.workspace = true name = "windmill_api" path = "src/lib.rs" -[[bin]] -name = "windmill_api" -path = "src/main.rs" - [features] enterprise = ["windmill-queue/enterprise"] diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index ca85300719..0bce8a9aab 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -13,6 +13,7 @@ use crate::{ users::{require_owner_of_path, Authed, OptAuthed}, variables::build_crypt, webhook_util::{WebhookMessage, WebhookShared}, + HTTP_CLIENT, }; use axum::{ extract::{Extension, Json, Path, Query}, @@ -21,7 +22,6 @@ use axum::{ }; use hyper::StatusCode; use magic_crypt::MagicCryptTrait; -use reqwest::Client; use serde::{Deserialize, Serialize}; use serde_json::{json, Map, Value}; use sha2::{Digest, Sha256}; @@ -368,12 +368,9 @@ async fn create_app( Ok((StatusCode::CREATED, app.path)) } -async fn list_hub_apps( - Authed { email, .. }: Authed, - Extension(http_client): Extension, -) -> JsonResult { +async fn list_hub_apps(Authed { email, .. }: Authed) -> JsonResult { let flows = list_elems_from_hub( - http_client, + &HTTP_CLIENT, "https://hub.windmill.dev/searchUiData?approved=true", &email, ) @@ -384,10 +381,9 @@ async fn list_hub_apps( pub async fn get_hub_app_by_id( Authed { email, .. }: Authed, Path(id): Path, - Extension(http_client): Extension, ) -> JsonResult { let value = http_get_from_hub( - http_client, + &HTTP_CLIENT, &format!("https://hub.windmill.dev/apps/{id}/json"), &email, false, diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 03ecf8bae7..747d8f2088 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -7,7 +7,6 @@ */ use hyper::StatusCode; -use reqwest::Client; use sql_builder::prelude::*; use axum::{ @@ -33,6 +32,7 @@ use crate::{ schedule::clear_schedule, users::{require_owner_of_path, Authed}, webhook_util::{WebhookMessage, WebhookShared}, + HTTP_CLIENT, }; pub fn workspaced_service() -> Router { @@ -111,12 +111,9 @@ async fn list_flows( Ok(Json(rows)) } -async fn list_hub_flows( - Authed { email, .. }: Authed, - Extension(http_client): Extension, -) -> JsonResult { +async fn list_hub_flows(Authed { email, .. }: Authed) -> JsonResult { let flows = list_elems_from_hub( - http_client, + &HTTP_CLIENT, "https://hub.windmill.dev/searchFlowData?approved=true", &email, ) @@ -145,10 +142,9 @@ async fn list_paths( pub async fn get_hub_flow_by_id( Authed { email, .. }: Authed, Path(id): Path, - Extension(http_client): Extension, ) -> JsonResult { let value = http_get_from_hub( - http_client, + &HTTP_CLIENT, &format!("https://hub.windmill.dev/flows/{id}/json"), &email, false, diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 7322811262..b076b42513 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -6,8 +6,6 @@ * LICENSE-AGPL for a copy of the license. */ -use std::sync::Arc; - use anyhow::Context; use axum::{ extract::{FromRequest, Json, Path, Query}, @@ -38,7 +36,7 @@ use crate::{ db::{UserDB, DB}, users::{require_owner_of_path, Authed}, variables::get_workspace_key, - BaseUrl, QueueLimitWaitResult, TimeoutWaitResult, + BASE_URL, }; pub fn workspaced_service() -> Router { @@ -925,16 +923,16 @@ pub async fn get_resume_urls( Extension(user_db): Extension, Path((w_id, job_id, resume_id)): Path<(String, Uuid, u32)>, Query(approver): Query, - Extension(base_url): Extension>, ) -> error::JsonResult { let key = get_workspace_key(&w_id, &mut user_db.begin(&authed).await?).await?; let signature = create_signature(key, job_id, resume_id, approver.approver.clone())?; - let base_url = base_url.0.clone(); let approver = approver .approver .as_ref() .map(|x| format!("?approver={}", encode(x))) .unwrap_or_else(String::new); + + let base_url = BASE_URL.as_str(); let res = ResumeUrls { approvalPage: format!( "{base_url}/approve/{w_id}/{job_id}/{resume_id}/{signature}{approver}" @@ -1315,18 +1313,26 @@ pub async fn check_queue_too_long(db: DB, queue_limit: Option) -> error::Re } Ok(()) } + +lazy_static::lazy_static! { + pub static ref QUEUE_LIMIT_WAIT_RESULT: Option = std::env::var("QUEUE_LIMIT_WAIT_RESULT") + .ok() + .and_then(|x| x.parse().ok()); + pub static ref TIMEOUT_WAIT_RESULT: i32 = std::env::var("TIMEOUT_WAIT_RESULT") + .ok() + .and_then(|x| x.parse().ok()) + .unwrap_or(20); +} pub async fn run_wait_result_job_by_path( authed: Authed, Extension(user_db): Extension, Extension(db): Extension, - Extension(timeout): Extension>, - Extension(queue_limit): Extension>, Path((w_id, script_path)): Path<(String, StripPath)>, Query(run_query): Query, headers: HeaderMap, Json(args): Json>>, ) -> error::JsonResult { - check_queue_too_long(db, queue_limit.0.or(run_query.queue_limit)).await?; + check_queue_too_long(db, QUEUE_LIMIT_WAIT_RESULT.or(run_query.queue_limit)).await?; let script_path = script_path.to_path(); let mut tx = user_db.clone().begin(&authed).await?; let job_payload = script_path_to_payload(script_path, &mut tx, &w_id).await?; @@ -1356,7 +1362,7 @@ pub async fn run_wait_result_job_by_path( run_wait_result( authed, Extension(user_db), - timeout.0, + *TIMEOUT_WAIT_RESULT, uuid, Path((w_id, script_path)), ) @@ -1367,7 +1373,6 @@ pub async fn run_wait_result_job_by_hash( authed: Authed, Extension(user_db): Extension, Extension(db): Extension, - Extension(timeout): Extension>, Path((w_id, script_hash)): Path<(String, ScriptHash)>, Query(run_query): Query, headers: HeaderMap, @@ -1403,7 +1408,7 @@ pub async fn run_wait_result_job_by_hash( run_wait_result( authed, Extension(user_db), - timeout.0, + *TIMEOUT_WAIT_RESULT, uuid, Path((w_id, script_hash)), ) @@ -1414,7 +1419,6 @@ pub async fn run_wait_result_flow_by_path( authed: Authed, Extension(user_db): Extension, Extension(db): Extension, - Extension(timeout): Extension>, Path((w_id, flow_path)): Path<(String, StripPath)>, Query(run_query): Query, headers: HeaderMap, @@ -1450,7 +1454,7 @@ pub async fn run_wait_result_flow_by_path( run_wait_result( authed, Extension(user_db), - timeout.0, + *TIMEOUT_WAIT_RESULT, uuid, Path((w_id, flow_path)), ) diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index f3a2bbcde0..b31c999bc7 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -6,15 +6,17 @@ * LICENSE-AGPL for a copy of the license. */ +use crate::oauth2::AllClients; use argon2::Argon2; use axum::{middleware::from_extractor, routing::get, Extension, Router}; use db::DB; use git_version::git_version; +use reqwest::Client; use std::{net::SocketAddr, sync::Arc}; use tower::ServiceBuilder; use tower_cookies::CookieManagerLayer; use tower_http::trace::TraceLayer; -use windmill_common::{error::to_anyhow, utils::rd_string}; +use windmill_common::utils::rd_string; use crate::{ db::UserDB, @@ -50,20 +52,32 @@ mod workspaces; pub const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); -pub struct BaseUrl(String); -pub struct IsSecure(bool); -pub struct CookieDomain(Option); -pub struct CloudHosted(bool); -pub struct ContentSecurityPolicy(String); -pub struct TimeoutWaitResult(i32); -pub struct QueueLimitWaitResult(Option); - pub use users::delete_expired_items_perdiodically; +lazy_static::lazy_static! { + pub static ref BASE_URL: String = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); + + + pub static ref COOKIE_DOMAIN: Option = std::env::var("COOKIE_DOMAIN").ok(); + + pub static ref SLACK_SIGNING_SECRET: Option = std::env::var("SLACK_SIGNING_SECRET") + .ok() + .map(|x| SlackVerifier::new(x).unwrap()); + + static ref IS_SECURE: bool = BASE_URL.starts_with("https://"); + + pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() + .user_agent("windmill/beta") + .build().unwrap(); + + pub static ref OAUTH_CLIENTS: AllClients = build_oauth_clients(&BASE_URL) + .map_err(|e| tracing::error!("Error building oauth clients: {}", e)) + .unwrap(); +} + pub async fn run_server( db: DB, addr: SocketAddr, - base_url: String, mut rx: tokio::sync::broadcast::Receiver<()>, ) -> anyhow::Result<()> { let user_db = UserDB::new(db.clone()); @@ -73,16 +87,7 @@ pub async fn run_server( std::env::var("SUPERADMIN_SECRET").ok(), )); let argon2 = Arc::new(Argon2::default()); - let basic_clients = Arc::new(build_oauth_clients(&base_url).await?); - let slack_verifier = Arc::new( - std::env::var("SLACK_SIGNING_SECRET") - .ok() - .map(|x| SlackVerifier::new(x).unwrap()), - ); - let http_client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build() - .map_err(to_anyhow)?; + let middleware_stack = ServiceBuilder::new() .layer( TraceLayer::new_for_http() @@ -93,21 +98,6 @@ pub async fn run_server( .layer(Extension(db.clone())) .layer(Extension(user_db)) .layer(Extension(auth_cache.clone())) - .layer(Extension(basic_clients)) - .layer(Extension(Arc::new(BaseUrl(base_url.to_string())))) - .layer(Extension(Arc::new(ContentSecurityPolicy( - std::env::var("SERVE_CSP").unwrap_or("".to_owned()), - )))) - .layer(Extension(Arc::new(CloudHosted( - std::env::var("CLOUD_HOSTED").is_ok(), - )))) - .layer(Extension(Arc::new(IsSecure( - base_url.starts_with("https://"), - )))) - .layer(Extension(Arc::new(CookieDomain( - std::env::var("COOKIE_DOMAIN").ok(), - )))) - .layer(Extension(http_client)) .layer(CookieManagerLayer::new()) .layer(Extension(WebhookShared::new(rx.resubscribe(), db.clone()))); // build our application with a route @@ -119,21 +109,7 @@ pub async fn run_server( "/w/:workspace_id", Router::new() .nest("/scripts", scripts::workspaced_service()) - .nest( - "/jobs", - jobs::workspaced_service() - .layer(Extension(Arc::new(TimeoutWaitResult( - std::env::var("TIMEOUT_WAIT_RESULT") - .ok() - .and_then(|x| x.parse().ok()) - .unwrap_or(20), - )))) - .layer(Extension(Arc::new(QueueLimitWaitResult( - std::env::var("QUEUE_LIMIT_WAIT_RESULT") - .ok() - .and_then(|x| x.parse().ok()), - )))), - ) + .nest("/jobs", jobs::workspaced_service()) .nest( "/users", users::workspaced_service().layer(Extension(argon2.clone())), @@ -174,10 +150,7 @@ pub async fn run_server( "/auth", users::make_unauthed_service().layer(Extension(argon2)), ) - .nest( - "/oauth", - oauth2::global_service().layer(Extension(slack_verifier)), - ) + .nest("/oauth", oauth2::global_service()) .route("/version", get(git_v)) .route("/openapi.yaml", get(openapi)), ) diff --git a/backend/windmill-api/src/main.rs b/backend/windmill-api/src/main.rs deleted file mode 100644 index e81e12861c..0000000000 --- a/backend/windmill-api/src/main.rs +++ /dev/null @@ -1,70 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2022 - * This file and its contents are licensed under the AGPLv3 License. - * Please see the included NOTICE for copyright information and - * LICENSE-AGPL for a copy of the license. - */ - -use std::net::SocketAddr; - -use anyhow::Ok; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - windmill_common::tracing_init::initialize_tracing(); - - let db = windmill_common::connect_db(true).await?; - - let num_workers = std::env::var("NUM_WORKERS") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_NUM_WORKERS as i32); - - let metrics_addr: Option = std::env::var("METRICS_ADDR") - .ok() - .map(|s| { - s.parse::() - .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) - .or_else(|_| s.parse::().map(Some)) - }) - .transpose()? - .flatten(); - - let server_mode = !std::env::var("DISABLE_SERVER") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false); - - if server_mode { - windmill_api::migrate_db(&db).await?; - } - - let (tx, rx) = tokio::sync::broadcast::channel::<()>(3); - let shutdown_signal = windmill_common::shutdown_signal(tx); - - let base_url = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); - - if server_mode || num_workers > 0 { - let addr = SocketAddr::from(([0, 0, 0, 0], 8000)); - - let server_f = async { - if server_mode { - windmill_api::run_server(db.clone(), addr, base_url, rx.resubscribe()).await?; - } - Ok(()) as anyhow::Result<()> - }; - - let metrics_f = async { - match metrics_addr { - Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe()) - .await - .map_err(anyhow::Error::from), - None => Ok(()), - } - }; - - futures::try_join!(shutdown_signal, server_f, metrics_f)?; - } - Ok(()) -} diff --git a/backend/windmill-api/src/main3.rs b/backend/windmill-api/src/main3.rs new file mode 100644 index 0000000000..bcfb107829 --- /dev/null +++ b/backend/windmill-api/src/main3.rs @@ -0,0 +1,69 @@ +// /* +// * Author: Ruben Fiszel +// * Copyright: Windmill Labs, Inc 2022 +// * This file and its contents are licensed under the AGPLv3 License. +// * Please see the included NOTICE for copyright information and +// * LICENSE-AGPL for a copy of the license. +// */ +// use std::net::SocketAddr; + +// use anyhow::Ok; + +// pub const DEFAULT_NUM_WORKERS: usize = 3; + +// #[tokio::main] +// async fn main() -> anyhow::Result<()> { +// windmill_common::tracing_init::initialize_tracing(); + +// let db = windmill_common::connect_db(true).await?; + +// let num_workers = std::env::var("NUM_WORKERS") +// .ok() +// .and_then(|x| x.parse::().ok()) +// .unwrap_or(DEFAULT_NUM_WORKERS as i32); + +// let metrics_addr: Option = std::env::var("METRICS_ADDR") +// .ok() +// .map(|s| { +// s.parse::() +// .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) +// .or_else(|_| s.parse::().map(Some)) +// }) +// .transpose()? +// .flatten(); + +// let server_mode = !std::env::var("DISABLE_SERVER") +// .ok() +// .and_then(|x| x.parse::().ok()) +// .unwrap_or(false); + +// if server_mode { +// windmill_api::migrate_db(&db).await?; +// } + +// let (tx, rx) = tokio::sync::broadcast::channel::<()>(3); +// let shutdown_signal = windmill_common::shutdown_signal(tx); + +// if server_mode || num_workers > 0 { +// let addr = SocketAddr::from(([0, 0, 0, 0], 8000)); + +// let server_f = async { +// if server_mode { +// windmill_api::run_server(db.clone(), addr, rx.resubscribe()).await?; +// } +// Ok(()) as anyhow::Result<()> +// }; + +// let metrics_f = async { +// match metrics_addr { +// Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe()) +// .await +// .map_err(anyhow::Error::from), +// None => Ok(()), +// } +// }; + +// futures::try_join!(shutdown_signal, server_f, metrics_f)?; +// } +// Ok(()) +// } diff --git a/backend/windmill-api/src/oauth2.rs b/backend/windmill-api/src/oauth2.rs index 6fd91c950a..154a6e8351 100644 --- a/backend/windmill-api/src/oauth2.rs +++ b/backend/windmill-api/src/oauth2.rs @@ -8,8 +8,6 @@ use std::{collections::HashMap, fmt::Debug}; -use std::sync::Arc; - use anyhow::Context; use axum::extract::FromRequestParts; use axum::http::request::Parts; @@ -29,7 +27,6 @@ use oauth2::{Client as OClient, *}; use reqwest::Client; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use sqlx::{Postgres, Transaction}; -use tokio::{fs::File, io::AsyncReadExt}; use tower_cookies::{Cookie, Cookies}; use windmill_audit::{audit_log, ActionKind}; use windmill_common::users::username_to_permissioned_as; @@ -41,15 +38,14 @@ use crate::{ db::{UserDB, DB}, variables::{build_crypt, encrypt}, workspaces::WorkspaceSettings, - BaseUrl, }; -use crate::{CookieDomain, IsSecure}; +use crate::{BASE_URL, HTTP_CLIENT, IS_SECURE, OAUTH_CLIENTS, SLACK_SIGNING_SECRET}; use windmill_common::error::{self, to_anyhow, Error}; use windmill_common::oauth2::*; use windmill_queue::JobPayload; -use std::str; +use std::{fs, str}; pub fn global_service() -> Router { Router::new() @@ -112,7 +108,7 @@ pub struct AllClients { pub slack: Option, } -pub async fn build_oauth_clients(base_url: &str) -> anyhow::Result { +pub fn build_oauth_clients(base_url: &str) -> anyhow::Result { let connect_configs = serde_json::from_str::>(include_str!( "../../oauth_connect.json" ))?; @@ -120,14 +116,12 @@ pub async fn build_oauth_clients(base_url: &str) -> anyhow::Result { "../../oauth_login.json" ))?; - let mut content = String::new(); let path = "./oauth.json"; - if std::path::Path::new(path).exists() { - let mut file = File::open(path).await?; - file.read_to_string(&mut content).await?; + let content = if std::path::Path::new(path).exists() { + fs::read_to_string(path).map_err(to_anyhow)? } else { - content.push_str("{}"); - } + "{}".to_string() + }; let oauths: HashMap = match serde_json::from_str::>(&content) { @@ -289,12 +283,10 @@ pub struct SlackBotToken { async fn connect( Path(client_name): Path, Query(query): Query>, - Extension(clients): Extension>, - Extension(is_secure): Extension>, cookies: Cookies, ) -> error::Result { let mut query = query.clone(); - let connects = &clients.connects; + let connects = &OAUTH_CLIENTS.connects; let scopes = query .get("scopes") .map(|x| x.split('+').map(|x| x.to_owned()).collect()); @@ -310,7 +302,7 @@ async fn connect( cookies, scopes, extra_params, - is_secure.0, + *IS_SECURE, ) } @@ -377,11 +369,9 @@ async fn delete_account( Ok(format!("Deleted account id {id}")) } -async fn list_logins( - Extension(clients): Extension>, -) -> error::JsonResult> { +async fn list_logins() -> error::JsonResult> { Ok(Json( - clients + OAUTH_CLIENTS .logins .keys() .map(|x| x.to_owned()) @@ -394,11 +384,9 @@ struct ScopesAndParams { scopes: Vec, extra_params: Option>, } -async fn list_connects( - Extension(clients): Extension>, -) -> error::JsonResult> { +async fn list_connects() -> error::JsonResult> { Ok(Json( - (&clients.connects) + (&OAUTH_CLIENTS.connects) .into_iter() .map(|(k, v)| { ( @@ -413,12 +401,8 @@ async fn list_connects( )) } -async fn connect_slack( - Extension(clients): Extension>, - Extension(is_secure): Extension>, - cookies: Cookies, -) -> error::Result { - let mut client = clients +async fn connect_slack(cookies: Cookies) -> error::Result { + let mut client = OAUTH_CLIENTS .slack .as_ref() .ok_or_else(|| error::Error::BadRequest("slack client not setup".to_string()))? @@ -429,7 +413,7 @@ async fn connect_slack( client.add_scope("commands"); let url = client.authorize_url(&state); - set_cookie(&state, cookies, is_secure.0); + set_cookie(&state, cookies, *IS_SECURE); Ok(Redirect::to(url.as_str())) } @@ -471,14 +455,9 @@ async fn disconnect_slack( Ok(format!("slack disconnected")) } -async fn login( - Extension(clients): Extension>, - Extension(is_secure): Extension>, - Path(client_name): Path, - cookies: Cookies, -) -> error::Result { - let clients = &clients.logins; - oauth_redirect(clients, client_name, cookies, None, None, is_secure.0) +async fn login(Path(client_name): Path, cookies: Cookies) -> error::Result { + let clients = &OAUTH_CLIENTS.logins; + oauth_redirect(clients, client_name, cookies, None, None, *IS_SECURE) } #[derive(Deserialize)] @@ -489,13 +468,11 @@ async fn refresh_token( authed: Authed, Path((w_id, id)): Path<(String, i32)>, Extension(user_db): Extension, - Extension(clients): Extension>, - Extension(http_client): Extension, Json(VariablePath { path }): Json, ) -> error::Result { let tx = user_db.begin(&authed).await?; - _refresh_token(tx, &path, w_id, id, clients, http_client).await?; + _refresh_token(tx, &path, w_id, id).await?; Ok(format!("Token at path {path} refreshed")) } @@ -505,8 +482,6 @@ pub async fn _refresh_token<'c>( path: &str, w_id: String, id: i32, - clients: Arc, - http_client: Client, ) -> error::Result { let account = sqlx::query!( "SELECT client, refresh_token FROM account WHERE workspace_id = $1 AND id = $2", @@ -516,14 +491,14 @@ pub async fn _refresh_token<'c>( .fetch_optional(&mut tx) .await?; let account = not_found_if_none(account, "Account", &id.to_string())?; - let client = (&clients + let client = (&OAUTH_CLIENTS .connects .get(&account.client) .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))? .client) .to_owned(); - let token = _exchange_token(client, &account.refresh_token, http_client).await; + let token = _exchange_token(client, &account.refresh_token).await; if let Err(token_err) = token { sqlx::query!( @@ -581,14 +556,10 @@ pub async fn _refresh_token<'c>( Ok(token_str) } -async fn _exchange_token( - client: OClient, - refresh_token: &str, - http_client: Client, -) -> Result { +async fn _exchange_token(client: OClient, refresh_token: &str) -> Result { let token_json = client .exchange_refresh_token(&RefreshToken::from(refresh_token.clone())) - .with_client(&http_client) + .with_client(&HTTP_CLIENT) .execute::() .await .map_err(to_anyhow)?; @@ -609,11 +580,9 @@ pub struct OAuthCallback { async fn connect_callback( cookies: Cookies, Path(client_name): Path, - Extension(clients): Extension>, - Extension(http_client): Extension, Json(callback): Json, ) -> error::JsonResult { - let client_w_scopes = &clients + let client_w_scopes = OAUTH_CLIENTS .connects .get(&client_name) .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))?; @@ -621,7 +590,7 @@ async fn connect_callback( let client = client_w_scopes.client.to_owned(); let extra_params = client_w_scopes.extra_params_callback.clone(); let token_response = - exchange_code::(callback, &cookies, client, &http_client, extra_params) + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, extra_params) .await?; Ok(Json(token_response)) @@ -632,17 +601,15 @@ async fn connect_slack_callback( authed: Authed, cookies: Cookies, Extension(user_db): Extension, - Extension(clients): Extension>, - Extension(http_client): Extension, Json(callback): Json, ) -> error::Result { - let client = clients + let client = OAUTH_CLIENTS .slack .as_ref() .ok_or_else(|| error::Error::BadRequest("slack client not setup".to_string()))? .to_owned(); let token = - exchange_code::(callback, &cookies, client, &http_client, None).await?; + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, None).await?; let mut tx = user_db.begin(&authed).await?; @@ -760,16 +727,14 @@ where async fn slack_command( SlackSig { sig, ts }: SlackSig, - Extension(slack_verifier): Extension>>, Extension(db): Extension, - Extension(base_url): Extension>, body: Bytes, ) -> error::Result { let form: SlackCommand = serde_urlencoded::from_bytes(&body) .map_err(|_| error::Error::BadRequest("invalid payload".to_string()))?; let body = String::from_utf8_lossy(&body); - if slack_verifier + if SLACK_SIGNING_SECRET .as_ref() .as_ref() .map(|sv| sv.verify(&ts, &body, &sig).ok()) @@ -827,7 +792,7 @@ async fn slack_command( ) .await?; tx.commit().await?; - let url = base_url.0.to_owned(); + let url = BASE_URL.to_owned(); return Ok(format!( "Job launched. See details at {url}/run/{uuid}?workspace={}", &settings.workspace_id @@ -852,31 +817,27 @@ pub struct UserInfo { async fn login_callback( Path(client_name): Path, cookies: Cookies, - Extension(clients): Extension>, Extension(db): Extension, - Extension(http_client): Extension, - Extension(is_secure): Extension>, - Extension(cookie_domain): Extension>, Json(callback): Json, ) -> error::Result { - let client_w_config = &clients + let client_w_config = &OAUTH_CLIENTS .logins .get(&client_name) .ok_or_else(|| error::Error::BadRequest("invalid client".to_string()))?; let client = client_w_config.client.to_owned(); let token_res = - exchange_code::(callback, &cookies, client, &http_client, None).await; + exchange_code::(callback, &cookies, client, &HTTP_CLIENT, None).await; if let Ok(token) = token_res { let token = &token.access_token.to_string(); let userinfo_url = client_w_config.userinfo_url.as_ref().ok_or_else(|| { Error::BadConfig(format!("Missing userinfo_url in client {client_name}")) })?; - let user = http_get_user_info::(&http_client, userinfo_url, token).await?; + let user = http_get_user_info::(&HTTP_CLIENT, userinfo_url, token).await?; let email = match client_name.as_str() { "github" => http_get_user_info::>( - &http_client, + &HTTP_CLIENT, "https://api.github.com/user/emails", token, ) @@ -912,15 +873,7 @@ async fn login_callback( if let Some((email, login_type, super_admin)) = login { let login_type = serde_json::json!(login_type); if login_type == client_name { - crate::users::create_session_token( - &email, - super_admin, - &mut tx, - cookies, - is_secure.0, - &cookie_domain.as_ref().0, - ) - .await?; + crate::users::create_session_token(&email, super_admin, &mut tx, cookies).await?; } else { return Err(error::Error::BadRequest(format!( "an user with the email associated to this login exists but with a different \ @@ -955,15 +908,7 @@ async fn login_callback( tx.commit().await?; invite_user_to_all_auto_invite_worspaces(&db, &email).await?; tx = db.begin().await?; - crate::users::create_session_token( - &email, - false, - &mut tx, - cookies, - is_secure.0, - &cookie_domain.as_ref().0, - ) - .await?; + crate::users::create_session_token(&email, false, &mut tx, cookies).await?; audit_log( &mut tx, &email, @@ -998,7 +943,7 @@ async fn login_callback( tx.commit().await?; if let Some(new_user_webhook) = NEW_USER_WEBHOOK.clone() { - let _ = http_client + let _ = HTTP_CLIENT .post(&new_user_webhook) .json(&serde_json::json!({"email" : &email, "event": "oauth_signup"})) .send() diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 444f9c2f3a..f9e559367e 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -6,7 +6,6 @@ * LICENSE-AGPL for a copy of the license. */ -use reqwest::Client; use sql_builder::prelude::*; use windmill_audit::{audit_log, ActionKind}; @@ -15,6 +14,7 @@ use crate::{ schedule::clear_schedule, users::{require_owner_of_path, Authed}, webhook_util::{WebhookMessage, WebhookShared}, + HTTP_CLIENT, }; use axum::{ extract::{Extension, Path, Query}, @@ -163,12 +163,9 @@ async fn list_scripts( Ok(Json(rows)) } -async fn list_hub_scripts( - Authed { email, .. }: Authed, - Extension(http_client): Extension, -) -> JsonResult { +async fn list_hub_scripts(Authed { email, .. }: Authed) -> JsonResult { let asks = list_elems_from_hub( - http_client, + &HTTP_CLIENT, "https://hub.windmill.dev/searchData?approved=true", &email, ) @@ -442,21 +439,16 @@ async fn create_script( Ok((StatusCode::CREATED, format!("{}", hash))) } -pub async fn get_hub_script_by_path( - authed: Authed, - Path(path): Path, - Extension(http_client): Extension, -) -> Result { - windmill_common::scripts::get_hub_script_by_path(&authed.email, path, http_client).await +pub async fn get_hub_script_by_path(authed: Authed, Path(path): Path) -> Result { + windmill_common::scripts::get_hub_script_by_path(&authed.email, path, &HTTP_CLIENT).await } pub async fn get_full_hub_script_by_path( Authed { email, .. }: Authed, Path(path): Path, - Extension(http_client): Extension, ) -> JsonResult { Ok(Json( - windmill_common::scripts::get_full_hub_script_by_path(&email, path, http_client).await?, + windmill_common::scripts::get_full_hub_script_by_path(&email, path, &HTTP_CLIENT).await?, )) } diff --git a/backend/windmill-api/src/static_assets.rs b/backend/windmill-api/src/static_assets.rs index 2b462a609c..ce9131ca8e 100644 --- a/backend/windmill-api/src/static_assets.rs +++ b/backend/windmill-api/src/static_assets.rs @@ -9,58 +9,41 @@ use axum::{ body::{self, BoxBody}, extract::OriginalUri, - http::{header, response::Builder, Response}, + http::{header, Response}, response::IntoResponse, - Extension, }; -use crate::{CloudHosted, ContentSecurityPolicy, IsSecure}; +use hyper::Uri; use mime_guess::mime; use rust_embed::RustEmbed; -use std::sync::Arc; // static_handler is a handler that serves static files from the -pub async fn static_handler( - Extension(is_secure): Extension>, - Extension(is_cloud_hosted): Extension>, - Extension(csp): Extension>, - OriginalUri(original_uri): OriginalUri, -) -> StaticFile { - let path = original_uri.path().trim_start_matches('/').to_string(); - StaticFile(path, is_secure.0, is_cloud_hosted.0, csp) +pub async fn static_handler(OriginalUri(original_uri): OriginalUri) -> StaticFile { + StaticFile(original_uri) } #[derive(RustEmbed)] #[folder = "../../frontend/build/"] struct Asset; -pub struct StaticFile( - pub String, - pub bool, - pub bool, - pub Arc, -); +pub struct StaticFile(Uri); impl IntoResponse for StaticFile { fn into_response(self) -> Response { - let path = self.0; - let can_set_security_headers = self.1 && self.2; - let csp = self.3; - serve_path(path, can_set_security_headers, csp) + let path = self.0.path().trim_start_matches('/'); + serve_path(path) } } -fn serve_path( - path: String, - can_set_security_headers: bool, - csp: Arc, -) -> Response { +const TWO_HUNDRED: &str = "200.html"; + +fn serve_path(path: &str) -> Response { if path.starts_with("api/") { return Response::builder() .status(404) .body(body::boxed(body::Empty::new())) .unwrap(); } - match Asset::get(path.as_str()) { + match Asset::get(path) { Some(content) => { let body = body::boxed(body::Full::from(content.data)); let mime = mime_guess::from_path(path).first_or_octet_stream(); @@ -75,26 +58,12 @@ fn serve_path( res = res.header(header::CACHE_CONTROL, "no-cache, no-store, must-revalidate"); } - if can_set_security_headers { - res = set_security_headers(res, csp); - } res.body(body).unwrap() } - None if path.as_str().starts_with("_app/") => Response::builder() + None if path.starts_with("_app/") => Response::builder() .status(404) .body(body::boxed(body::Empty::new())) .unwrap(), - None => serve_path("200.html".to_owned(), can_set_security_headers, csp), + None => serve_path(TWO_HUNDRED), } } - -fn set_security_headers(mut res: Builder, csp: Arc) -> Builder { - res = res.header("X-Frame-Options", "DENY"); - res = res.header("X-Content-Type-Options", "nosniff"); - - if !csp.0.is_empty() { - res = res.header("Content-Security-Policy", &csp.0); - } - - res -} diff --git a/backend/windmill-api/src/users.rs b/backend/windmill-api/src/users.rs index af9f5b1c55..7014d9d149 100644 --- a/backend/windmill-api/src/users.rs +++ b/backend/windmill-api/src/users.rs @@ -13,7 +13,7 @@ use crate::{ folders::get_folders_for_user, utils::require_super_admin, workspaces::invite_user_to_all_auto_invite_worspaces, - CookieDomain, IsSecure, + COOKIE_DOMAIN, IS_SECURE, }; use argon2::{password_hash::SaltString, Argon2, PasswordHash, PasswordHasher, PasswordVerifier}; use axum::{ @@ -721,14 +721,12 @@ async fn logout( Tokened { token }: Tokened, cookies: Cookies, Extension(db): Extension, - Extension(cookie_domain): Extension>, Query(LogoutQuery { rd }): Query, ) -> Result { let mut cookie = Cookie::new(COOKIE_NAME, ""); cookie.set_path(COOKIE_PATH); - let domain = cookie_domain.0.clone(); - if domain.is_some() { - cookie.set_domain(domain.clone().unwrap()); + if COOKIE_DOMAIN.is_some() { + cookie.set_domain(COOKIE_DOMAIN.clone().unwrap()); } cookies.remove(cookie); let mut tx = db.begin().await?; @@ -1565,8 +1563,6 @@ async fn login( cookies: Cookies, Extension(db): Extension, Extension(argon2): Extension>>, - Extension(is_secure): Extension>, - Extension(cookie_domain): Extension>, Json(Login { email, password }): Json, ) -> Result { let mut tx = db.begin().await?; @@ -1596,7 +1592,7 @@ async fn login( .execute(&mut tx) .await?; let mut c = Cookie::new("first_time", "1"); - if let Some(domain) = cookie_domain.as_ref().0.clone() { + if let Some(domain) = COOKIE_DOMAIN.as_ref() { c.set_domain(domain); } c.set_secure(false); @@ -1607,15 +1603,8 @@ async fn login( cookies.add(c); } - let token = create_session_token( - &email, - super_admin, - &mut tx, - cookies, - is_secure.0, - &cookie_domain.as_ref().0, - ) - .await?; + let token = create_session_token(&email, super_admin, &mut tx, cookies).await?; + tx.commit().await?; Ok(token) } @@ -1629,8 +1618,6 @@ pub async fn create_session_token<'c>( super_admin: bool, tx: &mut sqlx::Transaction<'c, sqlx::Postgres>, cookies: Cookies, - is_secure: bool, - domain: &Option, ) -> Result { let token = rd_string(30); sqlx::query!( @@ -1646,12 +1633,12 @@ pub async fn create_session_token<'c>( .execute(tx) .await?; let mut cookie = Cookie::new(COOKIE_NAME, token.clone()); - cookie.set_secure(is_secure); + cookie.set_secure(*IS_SECURE); cookie.set_same_site(Some(cookie::SameSite::Lax)); cookie.set_http_only(true); cookie.set_path(COOKIE_PATH); - if domain.is_some() { - cookie.set_domain(domain.clone().unwrap()); + if COOKIE_DOMAIN.is_some() { + cookie.set_domain(COOKIE_DOMAIN.clone().unwrap()); } let mut expire: OffsetDateTime = time::OffsetDateTime::now_utc(); expire += time::Duration::days(3); diff --git a/backend/windmill-api/src/variables.rs b/backend/windmill-api/src/variables.rs index 73f29c78a1..9436b88d08 100644 --- a/backend/windmill-api/src/variables.rs +++ b/backend/windmill-api/src/variables.rs @@ -6,14 +6,11 @@ * LICENSE-AGPL for a copy of the license. */ -use std::sync::Arc; - use crate::{ db::{UserDB, DB}, - oauth2::{AllClients, _refresh_token}, + oauth2::_refresh_token, users::{require_owner_of_path, Authed}, webhook_util::{WebhookMessage, WebhookShared}, - BaseUrl, }; /* * Author: Ruben Fiszel @@ -37,7 +34,6 @@ use windmill_common::{ }; use magic_crypt::{MagicCrypt256, MagicCryptTrait}; -use reqwest::Client; use serde::Deserialize; use sqlx::{Postgres, Transaction}; @@ -54,7 +50,6 @@ pub fn workspaced_service() -> Router { async fn list_contextual_variables( Path(w_id): Path, - Extension(base_url): Extension>, Authed { username, email, .. }: Authed, ) -> JsonResult> { Ok(Json( @@ -65,7 +60,6 @@ async fn list_contextual_variables( &username, "017e0ad5-f499-73b6-5488-92a61c5196dd", format!("u/{username}").as_str(), - &base_url.0, Some("u/user/script_path".to_string()), Some("017e0ad5-f499-73b6-5488-92a61c5196dd".to_string()), Some("u/user/encapsulating_flow_path".to_string()), @@ -111,8 +105,6 @@ async fn get_variable( Extension(user_db): Extension, Query(q): Query, Path((w_id, path)): Path<(String, StripPath)>, - Extension(clients): Extension>, - Extension(http_client): Extension, ) -> JsonResult { let path = path.to_path(); let mut tx = user_db.begin(&authed).await?; @@ -151,17 +143,7 @@ async fn get_variable( let value = variable.value.unwrap_or_else(|| "".to_string()); ListableVariable { value: if variable.is_expired.unwrap_or(false) && variable.account.is_some() { - Some( - _refresh_token( - tx, - &variable.path, - w_id, - variable.account.unwrap(), - clients, - http_client, - ) - .await?, - ) + Some(_refresh_token(tx, &variable.path, w_id, variable.account.unwrap()).await?) } else if !value.is_empty() && decrypt_secret { let mc = build_crypt(&mut tx, &w_id).await?; tx.commit().await?; diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index bde981dd54..c0b3520b99 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -6,7 +6,7 @@ * LICENSE-AGPL for a copy of the license. */ -use std::{str::FromStr, sync::Arc}; +use std::str::FromStr; use crate::{ db::{UserDB, DB}, @@ -14,7 +14,7 @@ use crate::{ resources::{Resource, ResourceType}, users::{Authed, WorkspaceInvite, NEW_USER_WEBHOOK}, utils::require_super_admin, - BaseUrl, + BASE_URL, HTTP_CLIENT, }; use axum::{ body::StreamBody, @@ -24,7 +24,6 @@ use axum::{ routing::{delete, get, post}, Json, Router, }; -use reqwest::Client; use stripe::CustomerId; use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ @@ -226,15 +225,14 @@ async fn stripe_checkout( authed: Authed, Path(w_id): Path, Query(plan): Query, - Extension(base_url): Extension>, ) -> Result { // #[cfg(feature = "enterprise")] { require_admin(authed.is_admin, &authed.username)?; let client = stripe::Client::new(std::env::var("STRIPE_KEY").expect("STRIPE_KEY")); - let success_rd = format!("{}/workspace_settings/checkout?success=true", base_url.0); - let failure_rd = format!("{}/workspace_settings/checkout?success=false", base_url.0); + let success_rd = format!("{}/workspace_settings/checkout?success=true", *BASE_URL); + let failure_rd = format!("{}/workspace_settings/checkout?success=false", *BASE_URL); let checkout_session = { let mut params = stripe::CreateCheckoutSession::new(&failure_rd, &success_rd); params.mode = Some(stripe::CheckoutSessionMode::Subscription); @@ -292,7 +290,6 @@ async fn stripe_portal( authed: Authed, Path(w_id): Path, Extension(db): Extension, - Extension(base_url): Extension>, ) -> Result { require_admin(authed.is_admin, &authed.username)?; let customer_id = sqlx::query_scalar!( @@ -303,7 +300,7 @@ async fn stripe_portal( .await? .ok_or_else(|| Error::InternalErr(format!("no customer id for workspace {}", w_id)))?; let client = stripe::Client::new(std::env::var("STRIPE_KEY").expect("STRIPE_KEY")); - let success_rd = format!("{}/workspace_settings?tab=premium", base_url.0); + let success_rd = format!("{}/workspace_settings?tab=premium", *BASE_URL); let portal_session = { let customer_id = CustomerId::from_str(&customer_id).unwrap(); let mut params = stripe::CreateBillingPortalSession::new(customer_id); @@ -926,7 +923,6 @@ pub async fn invite_user_to_all_auto_invite_worspaces(db: &DB, email: &str) -> R async fn invite_user( Authed { username, is_admin, .. }: Authed, Extension(db): Extension, - Extension(http_client): Extension, Path(w_id): Path, Json(nu): Json, ) -> Result<(StatusCode, String)> { @@ -949,7 +945,7 @@ async fn invite_user( tx.commit().await?; if let Some(new_user_webhook) = NEW_USER_WEBHOOK.clone() { - let _ = http_client + let _ = &HTTP_CLIENT .post(&new_user_webhook) .json(&serde_json::json!({"email" : &nu.email, "event": "new_invite"})) .send() diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index e35cc10691..946fb79d7f 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -41,3 +41,4 @@ hyper = { workspace = true, optional = true } tokio = { workspace = true, optional = true } reqwest = { workspace = true, optional = true } tracing-subscriber = { workspace = true, optional = true } +lazy_static.workspace = true \ No newline at end of file diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 046efb998c..105872d8f8 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -26,12 +26,13 @@ pub mod variables; #[cfg(feature = "tracing_init")] pub mod tracing_init; -pub const DEFAULT_NUM_WORKERS: usize = 3; -pub const DEFAULT_TIMEOUT: i32 = 300; -pub const DEFAULT_SLEEP_QUEUE: u64 = 50; pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50; pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 3; +lazy_static::lazy_static! { + pub static ref BASE_URL: String = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); +} + #[cfg(feature = "tokio")] pub async fn shutdown_signal(tx: tokio::sync::broadcast::Sender<()>) -> anyhow::Result<()> { use std::io; diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index b93efb3f4a..5a4a556cd3 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -210,7 +210,7 @@ pub fn to_hex_string(i: &i64) -> String { pub async fn get_hub_script_by_path( email: &str, path: StripPath, - http_client: reqwest::Client, + http_client: &reqwest::Client, ) -> crate::error::Result { use crate::{ error::{to_anyhow, Error}, @@ -239,7 +239,7 @@ pub async fn get_hub_script_by_path( pub async fn get_full_hub_script_by_path( email: &str, path: StripPath, - http_client: reqwest::Client, + http_client: &reqwest::Client, ) -> crate::error::Result { use crate::{ error::{to_anyhow, Error}, diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 7a79fe84ec..07253b0481 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -74,7 +74,7 @@ pub fn not_found_if_none>(opt: Option, kind: &str, name: U) #[cfg(feature = "reqwest")] pub async fn list_elems_from_hub( - http_client: reqwest::Client, + http_client: &reqwest::Client, url: &str, email: &str, ) -> Result { @@ -88,7 +88,7 @@ pub async fn list_elems_from_hub( #[cfg(feature = "reqwest")] pub async fn http_get_from_hub( - http_client: reqwest::Client, + http_client: &reqwest::Client, url: &str, email: &str, plain: bool, diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index a24f64e0d6..3262c05199 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -8,6 +8,8 @@ use serde::{Deserialize, Serialize}; +use crate::BASE_URL; + #[derive(Serialize, Clone)] pub struct ContextualVariable { @@ -66,7 +68,6 @@ pub fn get_reserved_variables( username: &str, job_id: &str, permissioned_as: &str, - base_url: &str, path: Option, flow_id: Option, flow_path: Option, @@ -114,7 +115,7 @@ pub fn get_reserved_variables( }, ContextualVariable { name: "WM_BASE_URL".to_string(), - value: base_url.to_string(), + value: BASE_URL.clone(), description: "base url of this instance".to_string(), }, ContextualVariable { diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d752f7c277..385be7191f 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -9,6 +9,7 @@ use std::{collections::HashMap, str::FromStr}; use anyhow::Context; +use reqwest::Client; use serde::{Deserialize, Serialize}; use sqlx::{Pool, Postgres, Transaction}; use tracing::{instrument, Instrument}; @@ -16,7 +17,7 @@ use ulid::Ulid; use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; use windmill_common::{ - error::{self, to_anyhow, Error}, + error::{self, Error}, flow_status::{FlowStatus, JobResult, MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL}, flows::{FlowModule, FlowModuleValue, FlowValue}, scripts::{get_full_hub_script_by_path, HubScript, ScriptHash, ScriptLang}, @@ -24,6 +25,10 @@ use windmill_common::{ }; lazy_static::lazy_static! { + pub static ref HTTP_CLIENT: Client = reqwest::ClientBuilder::new() + .user_agent("windmill/beta") + .build().unwrap(); + // TODO: these aren't synced, they should be moved into the queue abstraction once/if that happens. static ref QUEUE_PUSH_COUNT: prometheus::IntCounter = prometheus::register_int_counter!( "queue_push_count", @@ -406,7 +411,7 @@ pub async fn push<'c>( ) } JobPayload::ScriptHub { path } => { - let script = get_hub_script(path.clone(), email) + let script = get_hub_script(&HTTP_CLIENT, path.clone(), email) .await .context("error fetching hub script")?; ( @@ -610,11 +615,11 @@ pub fn canceled_job_to_result(job: &QueuedJob) -> serde_json::Value { serde_json::json!({"message": format!("Job canceled: {reason} by {canceler}"), "name": "Canceled", "reason": reason, "canceler": canceler}) } -pub async fn get_hub_script(path: String, email: &str) -> error::Result { - let client = reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .build() - .map_err(to_anyhow)?; +pub async fn get_hub_script( + client: &reqwest::Client, + path: String, + email: &str, +) -> error::Result { get_full_hub_script_by_path(email, StripPath(path), client) .await .map(|e| e) diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 0c7feb5fbc..aa52348db0 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -3,11 +3,10 @@ name = "windmill-worker" version.workspace = true authors.workspace = true edition.workspace = true -default-run = "worker" -[[bin]] -name = "worker" -path = "./src/main.rs" +[lib] +name = "windmill_worker" +path = "src/lib.rs" [features] default = [] diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index f6469df2f6..d4060acca6 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -34,10 +34,11 @@ pub async fn eval_timeout( env: Vec<(String, serde_json::Value)>, creds: Option, by_id: Option, - base_internal_url: String, + base_internal_url: &str, ) -> anyhow::Result { let expr2 = expr.clone(); let (sender, mut receiver) = oneshot::channel::(); + let base_internal_url: String = base_internal_url.to_string(); timeout( std::time::Duration::from_millis(2000), tokio::task::spawn_blocking(move || { @@ -346,7 +347,7 @@ mod tests { let code = "value.test + params.test"; let mut runtime = JsRuntime::new(RuntimeOptions::default()); - let res = eval(&mut runtime, code, env, None, None, "").await?; + let res = eval(&mut runtime, code, env, None, None, String::new().as_str()).await?; assert_eq!(res, json!(4)); Ok(()) } @@ -359,7 +360,7 @@ mod tests { multiline template`"; let mut runtime = JsRuntime::new(RuntimeOptions::default()); - let res = eval(&mut runtime, code, env, None, None, "").await?; + let res = eval(&mut runtime, code, env, None, None, String::new().as_str()).await?; assert_eq!(res, json!("my 5\nmultiline template")); Ok(()) } @@ -372,7 +373,7 @@ multiline template`"; ]; let code = r#"params.test"#; - let res = eval_timeout(code.to_string(), env, None, None, "".to_string()).await?; + let res = eval_timeout(code.to_string(), env, None, None, String::new().as_str()).await?; assert_eq!(res, json!(2)); Ok(()) } diff --git a/backend/windmill-worker/src/main.rs b/backend/windmill-worker/src/main.rs deleted file mode 100644 index f2b4cdb2ce..0000000000 --- a/backend/windmill-worker/src/main.rs +++ /dev/null @@ -1,138 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2022 - * This file and its contents are licensed under the AGPLv3 License. - * Please see the included NOTICE for copyright information and - * LICENSE-AGPL for a copy of the license. - */ - -use std::{net::SocketAddr, time::Duration}; - -use anyhow::Context; -use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; -use windmill_common::{ - error::{self, Error}, - utils::rd_string, -}; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - // dotenv().ok(); - - windmill_common::tracing_init::initialize_tracing(); - - let db = async { - let database_url = std::env::var("DATABASE_URL") - .map_err(|_| Error::BadConfig("DATABASE_URL env var is missing".to_string()))?; - - let max_connections = match std::env::var("DATABASE_CONNECTIONS") { - Ok(n) => n.parse::().context("invalid DATABASE_CONNECTIONS")?, - Err(_) => 10, - }; - - Ok::, error::Error>( - PgPoolOptions::new() - .max_connections(max_connections) - .max_lifetime(Duration::from_secs(30 * 60)) // 30 mins - .connect(&database_url) - .await - .map_err(|err| Error::ConnectingToDatabase(err.to_string()))?, - ) - } - .await?; - - let metrics_addr: Option = std::env::var("METRICS_ADDR") - .ok() - .map(|s| { - s.parse::() - .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) - .or_else(|_| s.parse::().map(Some)) - }) - .transpose()? - .flatten(); - - let (tx, rx) = tokio::sync::broadcast::channel::<()>(3); - let shutdown_signal = windmill_common::shutdown_signal(tx); - - let base_internal_url = - std::env::var("BASE_INTERNAL_URL").unwrap_or_else(|_| "http://localhost:8000".to_string()); - - let base_url = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string()); - - let timeout = std::env::var("TIMEOUT") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_TIMEOUT); - - let workers_f = async { - let sleep_queue = std::env::var("SLEEP_QUEUE") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(windmill_common::DEFAULT_SLEEP_QUEUE); - let disable_nuser = std::env::var("DISABLE_NUSER") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false); - let disable_nsjail = std::env::var("DISABLE_NSJAIL") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(true); - let keep_job_dir = std::env::var("KEEP_JOB_DIR") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(false); - let sync_bucket = std::env::var("S3_CACHE_BUCKET") - .ok() - .map(|e| Some(e)) - .unwrap_or(None); - - tracing::info!( - "DISABLE_NSJAIL: {disable_nsjail}, DISABLE_NUSER: {disable_nuser}, BASE_URL: \ - {base_url}, SLEEP_QUEUE: {sleep_queue}, TIMEOUT: \ - {timeout}, KEEP_JOB_DIR: {keep_job_dir}" - ); - let instance_name = rd_string(5); - - let ip = windmill_common::external_ip::get_ip() - .await - .unwrap_or_else(|e| { - tracing::warn!(error = e.to_string(), "failed to get external IP"); - "unretrievable IP".to_string() - }); - let worker_name = format!("dt-worker-{}-{}", &instance_name, rd_string(5)); - windmill_worker::run_worker( - &db.clone(), - timeout, - &instance_name, - worker_name, - 1, - 1, - &ip, - sleep_queue, - windmill_worker::WorkerConfig { - disable_nsjail, - disable_nuser, - base_internal_url, - base_url, - keep_job_dir, - }, - sync_bucket, - rx.resubscribe(), - ) - .await; - Ok(()) as anyhow::Result<()> - }; - - let metrics_f = async { - match metrics_addr { - Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe()) - .await - .map_err(anyhow::Error::from), - None => Ok(()), - } - }; - - futures::try_join!(shutdown_signal, workers_f, metrics_f)?; - - Ok(()) -} diff --git a/backend/windmill-worker/src/main2.rs b/backend/windmill-worker/src/main2.rs new file mode 100644 index 0000000000..fb79d6ba3e --- /dev/null +++ b/backend/windmill-worker/src/main2.rs @@ -0,0 +1,92 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2022 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +// use std::{net::SocketAddr, time::Duration}; + +// use anyhow::Context; +// use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; +// use windmill_common::{ +// error::{self, Error}, +// utils::rd_string, +// }; + +// #[tokio::main] +// async fn main() -> anyhow::Result<()> { +// // dotenv().ok(); + +// windmill_common::tracing_init::initialize_tracing(); + +// let db = async { +// let database_url = std::env::var("DATABASE_URL") +// .map_err(|_| Error::BadConfig("DATABASE_URL env var is missing".to_string()))?; + +// let max_connections = match std::env::var("DATABASE_CONNECTIONS") { +// Ok(n) => n.parse::().context("invalid DATABASE_CONNECTIONS")?, +// Err(_) => 10, +// }; + +// Ok::, error::Error>( +// PgPoolOptions::new() +// .max_connections(max_connections) +// .max_lifetime(Duration::from_secs(30 * 60)) // 30 mins +// .connect(&database_url) +// .await +// .map_err(|err| Error::ConnectingToDatabase(err.to_string()))?, +// ) +// } +// .await?; + +// let metrics_addr: Option = std::env::var("METRICS_ADDR") +// .ok() +// .map(|s| { +// s.parse::() +// .map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001)))) +// .or_else(|_| s.parse::().map(Some)) +// }) +// .transpose()? +// .flatten(); + +// let (tx, rx) = tokio::sync::broadcast::channel::<()>(3); +// let shutdown_signal = windmill_common::shutdown_signal(tx); + +// let workers_f = async { +// let instance_name = rd_string(5); + +// let ip = windmill_common::external_ip::get_ip() +// .await +// .unwrap_or_else(|e| { +// tracing::warn!(error = e.to_string(), "failed to get external IP"); +// "unretrievable IP".to_string() +// }); +// let worker_name = format!("dt-worker-{}-{}", &instance_name, rd_string(5)); +// windmill_worker::run_worker( +// &db.clone(), +// &instance_name, +// worker_name, +// 1, +// 1, +// &ip, +// rx.resubscribe(), +// ) +// .await; +// Ok(()) as anyhow::Result<()> +// }; + +// let metrics_f = async { +// match metrics_addr { +// Some(addr) => windmill_common::serve_metrics(addr, rx.resubscribe()) +// .await +// .map_err(anyhow::Error::from), +// None => Ok(()), +// } +// }; + +// futures::try_join!(shutdown_signal, workers_f, metrics_f)?; + +// Ok(()) +// } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 6a8b03701b..56f781575f 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -7,7 +7,6 @@ */ use const_format::concatcp; -use git_version::git_version; use itertools::Itertools; use lazy_static::lazy_static; use regex::Regex; @@ -24,7 +23,7 @@ use windmill_common::{ flows::{FlowModuleValue, FlowValue}, scripts::{ScriptHash, ScriptLang}, utils::rd_string, - variables, + variables, BASE_URL, }; use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob, CLOUD_HOSTED}; @@ -278,23 +277,95 @@ const NSJAIL_CONFIG_RUN_DENO_CONTENT: &str = include_str!("../nsjail/run.deno.co const RELATIVE_PYTHON_LOADER: &str = include_str!("../loader.py"); const GO_REQ_SPLITTER: &str = "//go.sum"; -const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version"); #[derive(Clone)] pub struct Metrics { pub worker_execution_failed: prometheus::IntCounter, } -#[derive(Clone, Debug)] -pub struct WorkerConfig { - pub base_internal_url: String, - pub base_url: String, - pub disable_nuser: bool, - pub disable_nsjail: bool, - pub keep_job_dir: bool, -} + +pub const DEFAULT_TIMEOUT: u16 = 300; +pub const DEFAULT_SLEEP_QUEUE: u64 = 50; lazy_static::lazy_static! { + + + + static ref SLEEP_QUEUE: u64 = std::env::var("SLEEP_QUEUE") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(DEFAULT_SLEEP_QUEUE); + + static ref DISABLE_NUSER: bool = std::env::var("DISABLE_NUSER") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(false); + + static ref DISABLE_NSJAIL: bool = std::env::var("DISABLE_NSJAIL") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(true); + + pub static ref KEEP_JOB_DIR: bool = std::env::var("KEEP_JOB_DIR") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(false); + + static ref S3_CACHE_BUCKET: Option = std::env::var("S3_CACHE_BUCKET") + .ok() + .map(|e| Some(e)) + .unwrap_or(None); + + static ref DENO_PATH: String = std::env::var("DENO_PATH").unwrap_or_else(|_| "/usr/bin/deno".to_string()); + static ref GO_PATH: String = std::env::var("GO_PATH").unwrap_or_else(|_| "/usr/bin/go".to_string()); + static ref PYTHON_PATH: String = + std::env::var("PYTHON_PATH").unwrap_or_else(|_| "/usr/local/bin/python3".to_string()); + static ref NSJAIL_PATH: String = std::env::var("NSJAIL_PATH").unwrap_or_else(|_| "nsjail".to_string()); + static ref PATH_ENV: String = std::env::var("PATH").unwrap_or_else(|_| String::new()); + static ref HOME_ENV: String = std::env::var("HOME").unwrap_or_else(|_| String::new()); + static ref PIP_INDEX_URL: Option = std::env::var("PIP_INDEX_URL").ok(); + static ref PIP_EXTRA_INDEX_URL: Option = std::env::var("PIP_EXTRA_INDEX_URL").ok(); + static ref PIP_TRUSTED_HOST: Option = std::env::var("PIP_TRUSTED_HOST").ok(); + static ref DENO_AUTH_TOKENS: String = std::env::var("DENO_AUTH_TOKENS") + .ok() + .map(|x| format!(";{x}")) + .unwrap_or_else(|| String::new()); + + + + + static ref DENO_FLAGS: Option> = std::env::var("DENO_FLAGS") + .ok() + .map(|x| x.split(' ').map(|x| x.to_string()).collect()); + + static ref PIP_LOCAL_DEPENDENCIES: Option> = { + let pip_local_dependencies = std::env::var("PIP_LOCAL_DEPENDENCIES") + .ok() + .map(|x| x.split(',').map(|x| x.to_string()).collect()); + if pip_local_dependencies == Some(vec!["".to_string()]) { + None + } else { + pip_local_dependencies + } + + }; + static ref WHITELIST_WORKSPACES: Option> = std::env::var("WHITELIST_WORKSPACES") + .ok() + .map(|x| x.split(',').map(|x| x.to_string()).collect()); + static ref BLACKLIST_WORKSPACES: Option> = std::env::var("BLACKLIST_WORKSPACES") + .ok() + .map(|x| x.split(',').map(|x| x.to_string()).collect()); + + static ref TAR_CACHE_RATE: i32 = std::env::var("TAR_CACHE_RATE") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(100); + + static ref ADDITIONAL_PYTHON_PATHS: Option> = std::env::var("ADDITIONAL_PYTHON_PATHS") + .ok() + .map(|x| x.split(':').map(|x| x.to_string()).collect()); + + static ref WORKER_STARTED: prometheus::IntGauge = prometheus::register_int_gauge!( "worker_started", "Total number of workers started." @@ -314,6 +385,17 @@ lazy_static::lazy_static! { "worker_uptime", "Total number of milliseconds since the worker has started" ); + + static ref TIMEOUT: u16 = std::env::var("TIMEOUT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(DEFAULT_TIMEOUT as u16); + static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT as u64); + + + static ref ZOMBIE_JOB_TIMEOUT: String = (*TIMEOUT as u32 * 5).to_string(); + + static ref SESSION_TOKEN_EXPIRY: i32 = (*TIMEOUT as i32) * 2; } //only matter if CLOUD_HOSTED @@ -324,18 +406,21 @@ const MAX_RESULT_SIZE: usize = 1024 * 1024 * 2; // 2MB #[tracing::instrument(level = "trace")] pub async fn run_worker( db: &Pool, - timeout: i32, worker_instance: &str, worker_name: String, i_worker: u64, - num_workers: u64, ip: &str, - sleep_queue: u64, - worker_config: WorkerConfig, - sync_bucket: Option, mut rx: tokio::sync::broadcast::Receiver<()>, + base_internal_url: &str, ) { - tracing::info!("Starting worker {worker_instance} {worker_name}, version: {GIT_VERSION}"); + + #[cfg(not(feature = "enterprise"))] + if !*DISABLE_NSJAIL { + tracing::warn!( + "NSJAIL to sandbox process in untrusted environments is an enterprise feature but allowed to be used for testing purposes" + ); + } + let start_time = Instant::now(); let worker_dir = format!("{TMP_DIR}/{worker_name}"); @@ -417,68 +502,7 @@ pub async fn run_worker( let mut jobs_executed = 0; - let deno_path = std::env::var("DENO_PATH").unwrap_or_else(|_| "/usr/bin/deno".to_string()); - let go_path = std::env::var("GO_PATH").unwrap_or_else(|_| "/usr/bin/go".to_string()); - let python_path = - std::env::var("PYTHON_PATH").unwrap_or_else(|_| "/usr/local/bin/python3".to_string()); - let nsjail_path = std::env::var("NSJAIL_PATH").unwrap_or_else(|_| "nsjail".to_string()); - let path_env = std::env::var("PATH").unwrap_or_else(|_| String::new()); - let home_env = std::env::var("HOME").unwrap_or_else(|_| String::new()); - let pip_index_url = std::env::var("PIP_INDEX_URL").ok(); - let pip_extra_index_url = std::env::var("PIP_EXTRA_INDEX_URL").ok(); - let pip_trusted_host = std::env::var("PIP_TRUSTED_HOST").ok(); - let deno_auth_tokens = std::env::var("DENO_AUTH_TOKENS") - .ok() - .map(|x| format!(";{x}")) - .unwrap_or_else(|| String::new()); - - - - let deno_flags = std::env::var("DENO_FLAGS") - .ok() - .map(|x| x.split(' ').map(|x| x.to_string()).collect()); - let pip_local_dependencies = std::env::var("PIP_LOCAL_DEPENDENCIES") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).collect()); - let whitelist_workspaces = std::env::var("WHITELIST_WORKSPACES") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).collect()); - let blacklist_workspaces = std::env::var("BLACKLIST_WORKSPACES") - .ok() - .map(|x| x.split(',').map(|x| x.to_string()).collect()); - - let pip_local_dependencies = if pip_local_dependencies == Some(vec!["".to_string()]) { - None - } else { - pip_local_dependencies - }; - - #[cfg(feature = "enterprise")] - let tar_cache_rate = std::env::var("TAR_CACHE_RATE") - .ok() - .and_then(|x| x.parse::().ok()) - .unwrap_or(100); - - let additional_python_paths = std::env::var("ADDITIONAL_PYTHON_PATHS") - .ok() - .map(|x| x.split(':').map(|x| x.to_string()).collect()); - - let envs = Envs { - deno_path, - go_path, - python_path, - nsjail_path, - path_env, - home_env, - pip_index_url, - pip_extra_index_url, - pip_trusted_host, - deno_flags, - deno_auth_tokens, - pip_local_dependencies, - additional_python_paths, - }; WORKER_STARTED.inc(); #[cfg(feature = "enterprise")] @@ -496,6 +520,8 @@ pub async fn run_worker( let (same_worker_tx, mut same_worker_rx) = mpsc::channel::(5); + tracing::info!(worker = %worker_name, "listening for jobs"); + loop { worker_busy.set(0); @@ -547,7 +573,7 @@ pub async fn run_worker( }, (job, timer) = { let timer = worker_pull_duration.start_timer(); - pull(&db, whitelist_workspaces.clone(), blacklist_workspaces.clone()).map(|x| (x, timer)) } => { + pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer)) } => { drop(timer); (false, job) }, @@ -617,12 +643,12 @@ pub async fn run_worker( &job.workspace_id, &job.permissioned_as, "ephemeral-script", - timeout * 2, + *SESSION_TOKEN_EXPIRY, &job.email, ) .await.expect("could not create job token"); tx.commit().await.expect("could not commit job token"); - let job_client = windmill_api_client::create_client(&worker_config.base_internal_url, token.clone()); + let job_client = windmill_api_client::create_client(base_internal_url, token.clone()); let is_flow = job.job_kind == JobKind::Flow || job.job_kind == JobKind::FlowPreview || job.job_kind == JobKind::FlowDependencies; if let Some(err) = handle_queued_job( @@ -630,15 +656,12 @@ pub async fn run_worker( db, &job_client, token, - timeout, &worker_name, &worker_dir, &job_dir, - &worker_config, metrics.clone(), - &envs, same_worker_tx.clone(), - &worker_config.base_internal_url, + base_internal_url ) .await .err() @@ -652,14 +675,13 @@ pub async fn run_worker( false, same_worker_tx.clone(), &worker_dir, - !worker_config.keep_job_dir, - &worker_config.base_internal_url, + base_internal_url ) .await; }; - if !worker_config.keep_job_dir && !(is_flow && same_worker) { + if !*KEEP_JOB_DIR && !(is_flow && same_worker) { let _ = tokio::fs::remove_dir_all(job_dir).await; } } @@ -668,7 +690,7 @@ pub async fn run_worker( let _timer = worker_sleep_duration .start_timer(); - tokio::time::sleep(Duration::from_millis(sleep_queue * num_workers)).await; + tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await; } Err(err) => { @@ -695,7 +717,6 @@ async fn handle_job_error( unrecoverable: bool, same_worker_tx: Sender, worker_dir: &str, - keep_job_dir: bool, base_internal_url: &str, ) { let err = match err { @@ -733,9 +754,8 @@ async fn handle_job_error( unrecoverable, same_worker_tx, worker_dir, - keep_job_dir, - base_internal_url, None, + base_internal_url ) .await; @@ -781,21 +801,6 @@ async fn insert_initial_ping( .expect("insert worker_ping initial value"); } -struct Envs { - deno_path: String, - go_path: String, - python_path: String, - nsjail_path: String, - path_env: String, - home_env: String, - pip_index_url: Option, - pip_extra_index_url: Option, - pip_trusted_host: Option, - deno_auth_tokens: String, - deno_flags: Option>, - pip_local_dependencies: Option>, - additional_python_paths: Option>, -} fn extract_error_value(log_lines: &str) -> serde_json::Value { return json!({"message": log_lines.to_string().trim().to_string(), "name": "ExecutionErr"}); @@ -806,15 +811,12 @@ async fn handle_queued_job( db: &sqlx::Pool, client: &windmill_api_client::Client, token: String, - timeout: i32, worker_name: &str, worker_dir: &str, job_dir: &str, - worker_config: &WorkerConfig, metrics: Metrics, - envs: &Envs, same_worker_tx: Sender, - base_internal_url: &str, + base_internal_url: &str ) -> windmill_common::error::Result<()> { if job.canceled { return Err(Error::JsonErr(canceled_job_to_result(&job)))?; @@ -833,7 +835,7 @@ async fn handle_queued_job( args, same_worker_tx, worker_dir, - base_internal_url, + base_internal_url ) .await?; } @@ -862,10 +864,10 @@ async fn handle_queued_job( logs.push_str(&format!("job {} on worker {}\n", &job.id, &worker_name)); let result = match job.job_kind { JobKind::Dependencies => { - handle_dependency_job(&job, &mut logs, job_dir, db, timeout, &envs).await + handle_dependency_job(&job, &mut logs, job_dir, db).await } JobKind::FlowDependencies => { - handle_flow_dependency_job(&job, &mut logs, job_dir, db, timeout, &envs) + handle_flow_dependency_job(&job, &mut logs, job_dir, db) .await .map(|()| Value::Null) } @@ -886,9 +888,7 @@ async fn handle_queued_job( job_dir, worker_dir, &mut logs, - timeout, - worker_config, - envs, + base_internal_url ) .await } @@ -911,9 +911,8 @@ async fn handle_queued_job( false, same_worker_tx.clone(), worker_dir, - worker_config.keep_job_dir, - &worker_config.base_internal_url, None, + base_internal_url ) .await?; } @@ -964,9 +963,8 @@ async fn handle_queued_job( false, same_worker_tx, worker_dir, - worker_config.keep_job_dir, - &worker_config.base_internal_url, None, + base_internal_url ) .await?; } @@ -1042,9 +1040,7 @@ async fn handle_code_execution_job( job_dir: &str, worker_dir: &str, logs: &mut String, - timeout: i32, - worker_config: &WorkerConfig, - envs: &Envs, + base_internal_url: &str ) -> error::Result { let (inner_content, requirements_o, language) = match job.job_kind { JobKind::Preview | JobKind::Script_Hub => ( @@ -1106,8 +1102,6 @@ mount {{ } Some(ScriptLang::Python3) => { handle_python_job( - worker_config, - envs, requirements_o, job_dir, worker_dir, @@ -1117,16 +1111,14 @@ mount {{ db, client, token, - timeout, &inner_content, &shared_mount, + base_internal_url ) .await } Some(ScriptLang::Deno) => { handle_deno_job( - worker_config, - envs, logs, job, db, @@ -1134,41 +1126,37 @@ mount {{ token, job_dir, &inner_content, - timeout, &shared_mount, requirements_o, + base_internal_url ) .await } Some(ScriptLang::Go) => { handle_go_job( - worker_config, - envs, logs, job, db, client, token, &inner_content, - timeout, job_dir, requirements_o, &shared_mount, + base_internal_url ) .await } Some(ScriptLang::Bash) => { handle_bash_job( - worker_config, - envs, logs, job, db, token, &inner_content, - timeout, job_dir, &shared_mount, + base_internal_url ) .await } @@ -1187,18 +1175,16 @@ mount {{ #[tracing::instrument(level = "trace", skip_all)] async fn handle_go_job( - WorkerConfig { base_internal_url, disable_nuser, disable_nsjail, base_url, .. }: &WorkerConfig, - Envs { nsjail_path, go_path, path_env, home_env, .. }: &Envs, logs: &mut String, job: &QueuedJob, db: &sqlx::Pool, client: &windmill_api_client::Client, token: String, inner_content: &str, - timeout: i32, job_dir: &str, requirements_o: Option, shared_mount: &str, + base_internal_url: &str, ) -> Result { //go does not like executing modules at temp root let job_dir = &format!("{job_dir}/go"); @@ -1225,8 +1211,6 @@ async fn handle_go_job( logs, job_dir, db, - timeout, - go_path, true, skip_go_mod, ) @@ -1321,79 +1305,77 @@ func Run(req Req) (interface{{}}, error){{ write_file(&format!("{job_dir}/inner"), "runner.go", &runner_content).await?; } } - let mut reserved_variables = get_reserved_variables(job, &token, &base_url, db).await?; + let mut reserved_variables = get_reserved_variables(job, &token, db).await?; reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); - let child = if !disable_nsjail { + let child = if !*DISABLE_NSJAIL { let _ = write_file( job_dir, "run.config.proto", &NSJAIL_CONFIG_RUN_GO_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", GO_CACHE_DIR) - .replace("{CLONE_NEWUSER}", &(!disable_nuser).to_string()) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount), ) .await?; - let build_go = Command::new(go_path) + let build_go = Command::new(GO_PATH.as_str()) .current_dir(job_dir) .env_clear() - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .env("GOPATH", GO_CACHE_DIR) - .env("HOME", home_env) + .env("HOME", HOME_ENV.as_str()) .args(vec!["build", "main.go"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(&job.id, db, logs, timeout, build_go, false).await?; + handle_child(&job.id, db, logs, build_go, false).await?; - Command::new(nsjail_path) + Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .args(vec!["--config", "run.config.proto", "--", "/tmp/go/main"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()? } else { - Command::new(go_path) + Command::new(GO_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .env("GOPATH", GO_CACHE_DIR) - .env("HOME", home_env) + .env("HOME", HOME_ENV.as_str()) .args(vec!["run", "main.go"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()? }; - handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; read_result(job_dir).await } #[tracing::instrument(level = "trace", skip_all)] async fn handle_bash_job( - WorkerConfig { base_internal_url, disable_nuser, disable_nsjail, base_url, .. }: &WorkerConfig, - Envs { nsjail_path, path_env, home_env, .. }: &Envs, logs: &mut String, job: &QueuedJob, db: &sqlx::Pool, token: String, content: &str, - timeout: i32, job_dir: &str, shared_mount: &str, + base_internal_url: &str, ) -> Result { logs.push_str("\n\n--- BASH CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; write_file(job_dir, "main.sh", content).await?; - let mut reserved_variables = get_reserved_variables(job, &token, &base_url, db).await?; + let mut reserved_variables = get_reserved_variables(job, &token, db).await?; reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); let hm = match job.args { @@ -1414,23 +1396,23 @@ async fn handle_bash_job( .collect::>(); let args = args_owned.iter().map(|s| &s[..]).collect::>(); - let child = if !disable_nsjail { + let child = if !*DISABLE_NSJAIL { let _ = write_file( job_dir, "run.config.proto", &NSJAIL_CONFIG_RUN_BASH_CONTENT .replace("{JOB_DIR}", job_dir) - .replace("{CLONE_NEWUSER}", &(!disable_nuser).to_string()) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount), ) .await?; let mut cmd_args = vec!["--config", "run.config.proto", "--", "/bin/sh", "main.sh"]; cmd_args.extend(args); - Command::new(nsjail_path) + Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .args(cmd_args) .stdout(Stdio::piped()) @@ -1443,15 +1425,15 @@ async fn handle_bash_job( .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) - .env("HOME", home_env) + .env("HOME", HOME_ENV.as_str()) .args(cmd_args) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()? }; - handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; //for now bash jobs have an empty result object Ok(serde_json::json!(logs .lines() @@ -1470,8 +1452,6 @@ fn capitalize(s: &str) -> String { #[tracing::instrument(level = "trace", skip_all)] async fn handle_deno_job( - WorkerConfig { base_internal_url, base_url, disable_nuser, disable_nsjail, .. }: &WorkerConfig, - Envs { nsjail_path, deno_path, path_env, deno_auth_tokens, deno_flags, .. }: &Envs, logs: &mut String, job: &QueuedJob, db: &sqlx::Pool, @@ -1479,9 +1459,9 @@ async fn handle_deno_job( token: String, job_dir: &str, inner_content: &String, - timeout: i32, shared_mount: &str, lockfile: Option, + base_internal_url: &str ) -> error::Result { logs.push_str("\n\n--- DENO CODE EXECUTION ---\n"); set_logs(logs, &job.id, db).await; @@ -1531,22 +1511,23 @@ run().catch(async (e) => {{ }}"# ); write_file(job_dir, "import_map.json", &import_map).await?; - let mut reserved_variables = get_reserved_variables(job, &token, &base_url, db).await?; + let mut reserved_variables = get_reserved_variables(job, &token, db).await?; reserved_variables.insert("RUST_LOG".to_string(), "info".to_string()); - let hostname_base = base_url.split("://").last().unwrap_or("localhost"); + let hostname_base = BASE_URL.split("://").last().unwrap_or("localhost"); let hostname_internal = base_internal_url.split("://").last().unwrap_or("localhost"); + let deno_auth_tokens_base = DENO_AUTH_TOKENS.as_str(); let deno_auth_tokens = - format!("{token}@{hostname_base};{token}@{hostname_internal}{deno_auth_tokens}",); + format!("{token}@{hostname_base};{token}@{hostname_internal}{deno_auth_tokens_base}",); let child = async { - Ok(if !disable_nsjail { + Ok(if !*DISABLE_NSJAIL { let _ = write_file( job_dir, "run.config.proto", &NSJAIL_CONFIG_RUN_DENO_CONTENT .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", DENO_CACHE_DIR) - .replace("{CLONE_NEWUSER}", &(!disable_nuser).to_string()) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount), ) .await?; @@ -1554,7 +1535,7 @@ run().catch(async (e) => {{ args.push("--config"); args.push("run.config.proto"); args.push("--"); - args.push(deno_path); + args.push(DENO_PATH.as_str()); args.push("run"); if lockfile.is_some() { args.push("--lock=/tmp/lock.json"); @@ -1562,7 +1543,7 @@ run().catch(async (e) => {{ args.push("--import-map"); args.push("/tmp/import_map.json"); args.push("--unstable"); - if let Some(deno_flags) = deno_flags { + if let Some(deno_flags) = DENO_FLAGS.as_ref() { for flag in deno_flags { args.push(flag); } @@ -1571,11 +1552,11 @@ run().catch(async (e) => {{ } args.push("/tmp/main.ts"); - Command::new(nsjail_path) + Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("DENO_AUTH_TOKENS", deno_auth_tokens) .env("BASE_INTERNAL_URL", base_internal_url) .args(args) @@ -1590,7 +1571,7 @@ run().catch(async (e) => {{ args.push("--import-map"); args.push(&import_map_path); args.push("--unstable"); - if let Some(deno_flags) = deno_flags { + if let Some(deno_flags) = DENO_FLAGS.as_ref() { for flag in deno_flags { args.push(flag); } @@ -1598,11 +1579,11 @@ run().catch(async (e) => {{ args.push("-A"); } args.push(&script_path); - Command::new(deno_path) + Command::new(DENO_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("DENO_DIR", DENO_CACHE_DIR) .env("DENO_AUTH_TOKENS", deno_auth_tokens) .env("BASE_INTERNAL_URL", base_internal_url) @@ -1615,7 +1596,7 @@ run().catch(async (e) => {{ } .instrument(trace_span!("create_deno_jail")) .await?; - handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; read_result(job_dir).await } @@ -1642,10 +1623,6 @@ lazy_static! { #[tracing::instrument(level = "trace", skip_all)] async fn handle_python_job( - WorkerConfig { base_internal_url, base_url, disable_nuser, disable_nsjail, .. }: &WorkerConfig, - envs @ Envs { - nsjail_path, python_path, path_env, additional_python_paths, .. - }: &Envs, requirements_o: Option, job_dir: &str, worker_dir: &str, @@ -1655,14 +1632,14 @@ async fn handle_python_job( db: &sqlx::Pool, client: &windmill_api_client::Client, token: String, - timeout: i32, inner_content: &String, shared_mount: &str, + base_internal_url: &str ) -> error::Result { create_dependencies_dir(job_dir).await; let mut additional_python_paths: Vec = - additional_python_paths.to_owned().unwrap_or_else(|| vec![]); + ADDITIONAL_PYTHON_PATHS.to_owned().unwrap_or_else(|| vec![]); let requirements = match requirements_o { Some(r) => r, @@ -1671,7 +1648,7 @@ async fn handle_python_job( if requirements.is_empty() { "".to_string() } else { - pip_compile(&job.id, &requirements, logs, job_dir, envs, db, timeout) + pip_compile(&job.id, &requirements, logs, job_dir, db) .await .map_err(|e| { Error::ExecutionErr(format!("pip compile failed: {}", e.to_string())) @@ -1681,14 +1658,14 @@ async fn handle_python_job( }; if requirements.len() > 0 { - if !disable_nsjail { + if !*DISABLE_NSJAIL { let _ = write_file( job_dir, "download.config.proto", &NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT .replace("{WORKER_DIR}", &worker_dir) .replace("{CACHE_DIR}", PIP_CACHE_DIR) - .replace("{CLONE_NEWUSER}", &(!disable_nuser).to_string()), + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()), ) .await?; } @@ -1698,12 +1675,9 @@ async fn handle_python_job( .split("\n") .filter(|x| !x.starts_with("--")) .collect(), - envs, job, logs, db, - timeout, - disable_nsjail.clone(), worker_name, job_dir, ) @@ -1814,9 +1788,9 @@ except Exception as e: ); write_file(job_dir, "main.py", &wrapper_content).await?; - let mut reserved_variables = get_reserved_variables(job, &token, &base_url, db).await?; + let mut reserved_variables = get_reserved_variables(job, &token, db).await?; let additional_python_paths_folders = additional_python_paths.iter().join(":"); - if !disable_nsjail { + if !*DISABLE_NSJAIL { let shared_deps = additional_python_paths .into_iter() .map(|pp| { @@ -1837,7 +1811,7 @@ mount {{ "run.config.proto", &NSJAIL_CONFIG_RUN_PYTHON3_CONTENT .replace("{JOB_DIR}", job_dir) - .replace("{CLONE_NEWUSER}", &(!disable_nuser).to_string()) + .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) .replace("{SHARED_MOUNT}", shared_mount) .replace("{SHARED_DEPENDENCIES}", shared_deps.as_str()) .replace( @@ -1857,19 +1831,19 @@ mount {{ "started python code execution {}", job.id ); - let child = if !disable_nsjail { - Command::new(nsjail_path) + let child = if !*DISABLE_NSJAIL { + Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) .env_clear() // inject PYTHONPATH here - for some reason I had to do it in nsjail conf .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .args(vec![ "--config", "run.config.proto", "--", - python_path, + PYTHON_PATH.as_str(), "-u", "/tmp/main.py", ]) @@ -1877,11 +1851,11 @@ mount {{ .stderr(Stdio::piped()) .spawn()? } else { - Command::new(python_path) + Command::new(PYTHON_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(reserved_variables) - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .env("BASE_INTERNAL_URL", base_internal_url) .args(vec!["-u", "main.py"]) .stdout(Stdio::piped()) @@ -1889,7 +1863,7 @@ mount {{ .spawn()? }; - handle_child(&job.id, db, logs, timeout, child, !disable_nsjail).await?; + handle_child(&job.id, db, logs, child, !*DISABLE_NSJAIL).await?; read_result(job_dir).await } @@ -1919,8 +1893,6 @@ async fn handle_dependency_job( logs: &mut String, job_dir: &str, db: &sqlx::Pool, - timeout: i32, - envs: &Envs, ) -> error::Result { let content = capture_dependency_job( &job.id, @@ -1935,9 +1907,7 @@ async fn handle_dependency_job( .unwrap_or_else(|| "no raw code"), logs, job_dir, - db, - timeout, - envs, + db ) .await; match content { @@ -1972,8 +1942,6 @@ async fn handle_flow_dependency_job( logs: &mut String, job_dir: &str, db: &sqlx::Pool, - timeout: i32, - envs: &Envs, ) -> error::Result<()> { let path = job.script_path.clone().ok_or_else(|| { error::Error::InternalErr( @@ -2004,8 +1972,6 @@ async fn handle_flow_dependency_job( logs, job_dir, db, - timeout, - envs, ) .await; match new_lock { @@ -2068,67 +2034,64 @@ async fn handle_flow_dependency_job( Ok(()) } -#[cfg(not(feature = "deno-lock"))] -async fn generate_deno_lock( - _job_id: &Uuid, - _code: &str, - _logs: &mut String, - _job_dir: &str, - _db: &sqlx::Pool, - _timeout: i32, - _envs: &Envs, -) -> error::Result { - Ok(String::new()) -} +// #[cfg(not(feature = "deno-lock"))] +// async fn generate_deno_lock( +// _job_id: &Uuid, +// _code: &str, +// _logs: &mut String, +// _job_dir: &str, +// _db: &sqlx::Pool, +// _timeout: i32, +// ) -> error::Result { +// Ok(String::new()) +// } -#[cfg(feature = "deno-lock")] -async fn generate_deno_lock( - job_id: &Uuid, - code: &str, - logs: &mut String, - job_dir: &str, - db: &sqlx::Pool, - timeout: i32, - Envs { deno_path, .. }: &Envs, -) -> error::Result { - let _ = write_file(job_dir, "main.ts", code).await?; +// #[cfg(feature = "deno-lock")] +// async fn generate_deno_lock( +// job_id: &Uuid, +// code: &str, +// logs: &mut String, +// job_dir: &str, +// db: &sqlx::Pool, +// timeout: i32, +// ) -> error::Result { +// let _ = write_file(job_dir, "main.ts", code).await?; - let child = Command::new(deno_path) - .current_dir(job_dir) - .args(vec![ - "cache", - "--unstable", - "--lock=lock.json", - "--lock-write", - "main.ts", - ]) - .env("NO_COLOR", "1") - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn()?; +// let child = Command::new(deno_path) +// .current_dir(job_dir) +// .args(vec![ +// "cache", +// "--unstable", +// "--lock=lock.json", +// "--lock-write", +// "main.ts", +// ]) +// .env("NO_COLOR", "1") +// .stdout(Stdio::piped()) +// .stderr(Stdio::piped()) +// .spawn()?; - handle_child(job_id, db, logs, timeout, * child).await?; +// handle_child(job_id, db, logs, timeout, child).await?; + +// let path_lock = format!("{job_dir}/lock.json"); +// let mut file = File::open(path_lock).await?; +// let mut req_content = "".to_string(); +// file.read_to_string(&mut req_content).await?; +// Ok(req_content) +// } - let path_lock = format!("{job_dir}/lock.json"); - let mut file = File::open(path_lock).await?; - let mut req_content = "".to_string(); - file.read_to_string(&mut req_content).await?; - Ok(req_content) -} async fn capture_dependency_job( job_id: &Uuid, job_language: &ScriptLang, job_raw_code: &str, logs: &mut String, job_dir: &str, - db: &sqlx::Pool, - timeout: i32, - envs: &Envs, + db: &sqlx::Pool ) -> error::Result { match job_language { ScriptLang::Python3 => { create_dependencies_dir(job_dir).await; - pip_compile(job_id, job_raw_code, logs, job_dir, envs, db, timeout).await + pip_compile(job_id, job_raw_code, logs, job_dir, db ).await } ScriptLang::Go => { install_go_dependencies( @@ -2137,15 +2100,14 @@ async fn capture_dependency_job( logs, job_dir, db, - timeout, - &envs.go_path, false, false, ) .await } ScriptLang::Deno => { - generate_deno_lock(job_id, job_raw_code, logs, job_dir, db, timeout, &envs).await + Ok(String::new()) + // generate_deno_lock(job_id, job_raw_code, logs, job_dir, db, timeout).await } ScriptLang::Bash => Ok("".to_owned()), } @@ -2156,22 +2118,13 @@ async fn pip_compile( requirements: &str, logs: &mut String, job_dir: &str, - Envs { - pip_extra_index_url, - pip_index_url, - pip_trusted_host, - - pip_local_dependencies, - .. - }: &Envs, db: &Pool, - timeout: i32, ) -> error::Result { logs.push_str(&format!("\nresolving dependencies...")); set_logs(logs, job_id, db).await; logs.push_str(&format!("\ncontent of requirements:\n{}", requirements)); let file = "requirements.in"; - let requirements = if let Some(pip_local_dependencies) = pip_local_dependencies { + let requirements = if let Some(pip_local_dependencies) = PIP_LOCAL_DEPENDENCIES.as_ref() { let deps = pip_local_dependencies.clone(); requirements .lines() @@ -2183,13 +2136,13 @@ async fn pip_compile( write_file(job_dir, file, &requirements).await?; let mut args = vec!["-q", "--no-header", file, "--resolver=backtracking"]; - if let Some(url) = pip_extra_index_url { + if let Some(url) = PIP_EXTRA_INDEX_URL.as_ref() { args.extend(["--extra-index-url", url]); } - if let Some(url) = pip_index_url { + if let Some(url) = PIP_INDEX_URL.as_ref() { args.extend(["--index-url", url]); } - if let Some(host) = pip_trusted_host { + if let Some(host) = PIP_TRUSTED_HOST.as_ref() { args.extend(["--trusted-host", host]); } let child = Command::new("pip-compile") @@ -2198,7 +2151,7 @@ async fn pip_compile( .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, timeout, child, false) + handle_child(job_id, db, logs, child, false) .await .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?; let path_lock = format!("{job_dir}/requirements.txt"); @@ -2219,8 +2172,6 @@ async fn install_go_dependencies( logs: &mut String, job_dir: &str, db: &sqlx::Pool, - timeout: i32, - go_path: &str, preview: bool, skip_go_mod: bool, ) -> error::Result { @@ -2233,15 +2184,15 @@ async fn install_go_dependencies( .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, timeout, child, false).await?; + handle_child(job_id, db, logs, child, false).await?; } - let child = Command::new(go_path) + let child = Command::new(GO_PATH.as_str()) .current_dir(job_dir) .args(vec!["mod", "tidy"]) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; - handle_child(job_id, db, logs, timeout, child, false) + handle_child(job_id, db, logs, child, false) .await .map_err(|e| Error::ExecutionErr(format!("Lock file generation failed: {e:?}")))?; @@ -2280,7 +2231,6 @@ async fn gen_go_mymod(code: &str, job_dir: &str) -> error::Result<()> { async fn get_reserved_variables( job: &QueuedJob, token: &str, - base_url: &str, db: &sqlx::Pool, ) -> Result, Error> { let flow_path = if let Some(uuid) = job.parent_job { @@ -2299,7 +2249,7 @@ async fn get_reserved_variables( &job.created_by, &job.id.to_string(), &job.permissioned_as, - base_url, + job.script_path.clone(), job.parent_job.map(|x| x.to_string()), flow_path, @@ -2346,11 +2296,9 @@ async fn handle_child( job_id: &Uuid, db: &Pool, logs: &mut String, - timeout: i32, mut child: Child, nsjail: bool, ) -> error::Result<()> { - let timeout = Duration::from_secs(u64::try_from(timeout).expect("invalid timeout")); let update_job_interval = Duration::from_millis(500); let write_logs_delay = Duration::from_millis(500); @@ -2415,7 +2363,7 @@ async fn handle_child( result = child.wait() => return result.map(Ok), Ok(()) = too_many_logs.changed() => KillReason::TooManyLogs, _ = update_job => KillReason::Cancelled, - _ = sleep(timeout) => KillReason::Timeout, + _ = sleep(*TIMEOUT_DURATION) => KillReason::Timeout, }; tx.send(()).await.expect("rx should never be dropped"); drop(tx); @@ -2431,7 +2379,7 @@ async fn handle_child( WHERE id = $2 "#, ) - .bind(format!("duration > {}", timeout.as_secs())) + .bind(format!("duration > {}", TIMEOUT_DURATION.as_secs())) .bind(job_id) .execute(&db) .await @@ -2661,12 +2609,11 @@ async fn append_logs(job_id: uuid::Uuid, logs: impl AsRef, db: impl Borrow< pub async fn handle_zombie_jobs_periodically( db: &Pool, - timeout: i32, - base_url: &str, mut rx: tokio::sync::broadcast::Receiver<()>, + base_internal_url: &str, ) { loop { - handle_zombie_jobs(db, timeout, base_url).await; + handle_zombie_jobs(db, base_internal_url).await; tokio::select! { _ = tokio::time::sleep(Duration::from_secs(60)) => (), @@ -2678,10 +2625,10 @@ pub async fn handle_zombie_jobs_periodically( } } -async fn handle_zombie_jobs(db: &Pool, timeout: i32, base_url: &str) { +async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { let restarted = sqlx::query!( "UPDATE queue SET running = false WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping", - (timeout * 5).to_string(), + *ZOMBIE_JOB_TIMEOUT, JobKind::Flow: JobKind, ) .fetch_all(db) @@ -2702,7 +2649,7 @@ async fn handle_zombie_jobs(db: &Pool, timeout: i32, base_url: &str) { let timeouts = sqlx::query_as::<_, QueuedJob>( "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = true", ) - .bind((timeout * 5).to_string()) + .bind(ZOMBIE_JOB_TIMEOUT.as_str()) .bind(JobKind::Flow) .fetch_all(db) .await @@ -2726,13 +2673,13 @@ async fn handle_zombie_jobs(db: &Pool, timeout: i32, base_url: &str) { &job.workspace_id, &job.permissioned_as, "ephemeral-zombie-jobs", - timeout * 2, + *SESSION_TOKEN_EXPIRY, &job.email, ) .await .expect("could not create job token"); tx.commit().await.expect("could not commit job token"); - let client = windmill_api_client::create_client(base_url, token.clone()); + let client = windmill_api_client::create_client(base_internal_url, token.clone()); let _ = handle_job_error( db, @@ -2743,9 +2690,7 @@ async fn handle_zombie_jobs(db: &Pool, timeout: i32, base_url: &str) { true, same_worker_tx_never_used, "", - true, - &std::env::var("BASE_INTERNAL_URL") - .unwrap_or_else(|_| "http://localhost:8000".to_string()), + base_internal_url, ) .await; } @@ -2753,34 +2698,22 @@ async fn handle_zombie_jobs(db: &Pool, timeout: i32, base_url: &str) { async fn handle_python_reqs( requirements: Vec<&str>, - Envs { - python_path, - path_env, - pip_index_url, - pip_extra_index_url, - pip_trusted_host, - nsjail_path, - - .. - }: &Envs, job: &QueuedJob, logs: &mut String, db: &sqlx::Pool, - timeout: i32, - disable_nsjail: bool, worker_name: &str, job_dir: &str, ) -> error::Result> { let mut req_paths: Vec = vec![]; - let mut vars = vec![("PATH", path_env)]; - if !disable_nsjail { - if let Some(url) = pip_extra_index_url { + let mut vars = vec![("PATH", PATH_ENV.as_str())]; + if !*DISABLE_NSJAIL { + if let Some(url) = PIP_EXTRA_INDEX_URL.as_ref() { vars.push(("EXTRA_INDEX_URL", url)); } - if let Some(url) = pip_index_url { + if let Some(url) = PIP_INDEX_URL.as_ref() { vars.push(("INDEX_URL", url)); } - if let Some(host) = pip_trusted_host { + if let Some(host) = PIP_TRUSTED_HOST.as_ref() { vars.push(("TRUSTED_HOST", host)); } }; @@ -2803,7 +2736,7 @@ async fn handle_python_reqs( "started setup python dependencies" ); - let child = if !disable_nsjail { + let child = if !*DISABLE_NSJAIL { tracing::info!( worker_name = %worker_name, job_id = %job.id, @@ -2814,7 +2747,7 @@ async fn handle_python_reqs( let req = req.to_string(); vars.push(("REQ", &req)); vars.push(("TARGET", &venv_p)); - Command::new(nsjail_path) + Command::new(NSJAIL_PATH.as_str()) .current_dir(job_dir) .env_clear() .envs(vars) @@ -2837,25 +2770,25 @@ async fn handle_python_reqs( "-t", venv_p.as_str(), ]; - if let Some(url) = pip_extra_index_url { - args.extend(["--extra-index-url", &url]); + if let Some(url) = PIP_EXTRA_INDEX_URL.as_ref() { + args.extend(["--extra-index-url", url]); } - if let Some(url) = pip_index_url { - args.extend(["--index-url", &url]); + if let Some(url) = PIP_INDEX_URL.as_ref() { + args.extend(["--index-url", url]); } - if let Some(host) = pip_trusted_host { + if let Some(host) = PIP_TRUSTED_HOST.as_ref() { args.extend(["--trusted-host", &host]); } - Command::new(python_path) + Command::new(PYTHON_PATH.as_str()) .env_clear() - .env("PATH", path_env) + .env("PATH", PATH_ENV.as_str()) .args(args) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()? }; - let child = handle_child(&job.id, db, logs, timeout, child, false).await; + let child = handle_child(&job.id, db, logs, child, false).await; tracing::info!( worker_name = %worker_name, job_id = %job.id, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 4df7866f5d..9445452cc5 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -12,7 +12,7 @@ use std::time::Duration; use crate::jobs::{add_completed_job, add_completed_job_error, schedule_again_if_scheduled}; use crate::js_eval::{eval_timeout, EvalCreds, IdContext}; -use crate::worker; +use crate::{worker, KEEP_JOB_DIR}; use anyhow::Context; use async_recursion::async_recursion; use dyn_iter::DynIter; @@ -50,9 +50,8 @@ pub async fn update_flow_status_after_job_completion( unrecoverable: bool, same_worker_tx: Sender, worker_dir: &str, - keep_job_dir: bool, - base_internal_url: &str, stop_early_override: Option, + base_internal_url: &str, ) -> error::Result<()> { tracing::debug!("UPDATE FLOW STATUS: {flow:?} {success} {result:?} {w_id}"); @@ -121,7 +120,7 @@ pub async fn update_flow_status_after_job_completion( let stop_early = success && if let Some(expr) = r.stop_early_expr.clone() { - compute_bool_from_expr(expr, &r.args, result.clone(), base_internal_url, None, None) + compute_bool_from_expr(expr, &r.args, result.clone(), None, None, base_internal_url) .await? } else { false @@ -481,7 +480,7 @@ pub async fn update_flow_status_after_job_completion( }; if done { - if flow_job.same_worker && !keep_job_dir { + if flow_job.same_worker && !*KEEP_JOB_DIR { let _ = tokio::fs::remove_dir_all(format!("{worker_dir}/{}", flow_job.id)).await; } @@ -498,13 +497,12 @@ pub async fn update_flow_status_after_job_completion( false, same_worker_tx.clone(), worker_dir, - keep_job_dir, - base_internal_url, if stop_early { Some(skip_if_stop_early) } else { None }, + base_internal_url, ) .await?); } @@ -583,9 +581,9 @@ async fn compute_bool_from_expr( expr: String, flow_args: &Option, result: serde_json::Value, - base_internal_url: &str, by_id: Option, creds: Option, + base_internal_url: &str, ) -> error::Result { let flow_input = flow_args.clone().unwrap_or_else(|| json!({})); match eval_timeout( @@ -598,7 +596,7 @@ async fn compute_bool_from_expr( .into(), creds, by_id, - base_internal_url.to_string(), + base_internal_url, ) .await? { @@ -722,7 +720,7 @@ async fn transform_input( context, Some(EvalCreds { workspace: workspace.to_string(), token: token.to_string() }), Some(by_id.clone()), - base_internal_url.to_string(), + base_internal_url, ) .await .map_err(|e| { @@ -769,8 +767,8 @@ pub async fn handle_flow( client, last_result, same_worker_tx, - base_internal_url, worker_dir, + base_internal_url, ) .await?; Ok(()) @@ -786,8 +784,8 @@ async fn push_next_flow_job( client: &windmill_api_client::Client, mut last_result: serde_json::Value, same_worker_tx: Sender, - base_internal_url: &str, worker_dir: &str, + base_internal_url: &str, ) -> error::Result<()> { let mut i = usize::try_from(status.step) .with_context(|| format!("invalid module index {}", status.step))?; @@ -816,9 +814,8 @@ async fn push_next_flow_job( true, same_worker_tx, worker_dir, - false, - base_internal_url, None, + base_internal_url, ) .await; } @@ -858,7 +855,7 @@ async fn push_next_flow_job( .into(), None, None, - "".to_string(), + base_internal_url, ) .await .map_err(|e| { @@ -1162,8 +1159,8 @@ async fn push_next_flow_job( &status, &status_module, last_result.clone(), - base_internal_url, previous_id, + base_internal_url, ) .await?; tx.commit().await?; @@ -1504,8 +1501,8 @@ async fn compute_next_flow_transform<'c>( status: &FlowStatus, status_module: &FlowStatusModule, last_result: serde_json::Value, - base_internal_url: &str, previous_id: String, + base_internal_url: &str, ) -> error::Result<(sqlx::Transaction<'c, sqlx::Postgres>, NextFlowTransform)> { match &module.value { FlowModuleValue::Identity => Ok(( @@ -1701,12 +1698,12 @@ async fn compute_next_flow_transform<'c>( b.expr.to_string(), &flow_job.args, last_result.clone(), - base_internal_url, Some(idcontext.clone()), Some(EvalCreds { workspace: flow_job.workspace_id.clone(), token: token.to_string(), }), + base_internal_url, ) .await?; @@ -1920,7 +1917,7 @@ where vars(), Some(EvalCreds { workspace, token }), by_id, - base_internal_url.to_string(), + base_internal_url, ) .await } diff --git a/frontend/src/routes/(root)/(logged)/+layout.svelte b/frontend/src/routes/(root)/(logged)/+layout.svelte index bb3dbd65c7..80d20d0441 100644 --- a/frontend/src/routes/(root)/(logged)/+layout.svelte +++ b/frontend/src/routes/(root)/(logged)/+layout.svelte @@ -19,7 +19,6 @@ import FavoriteMenu from '$lib/components/sidebar/FavoriteMenu.svelte' OpenAPI.WITH_CREDENTIALS = true - let menuOpen = false let isCollapsed = false let userSettings: UserSettings