diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 7c9a4f373c..acfe670ca9 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15483,6 +15483,7 @@ dependencies = [ "magic-crypt", "mail-send", "object_store", + "once_cell", "openidconnect", "opentelemetry", "opentelemetry-appender-tracing", diff --git a/backend/src/main.rs b/backend/src/main.rs index a619572f50..699250f1fa 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -82,11 +82,11 @@ static GLOBAL: Jemalloc = Jemalloc; use windmill_common::global_settings::OBJECT_STORE_CONFIG_SETTING; use windmill_worker::{ - get_hub_script_content_and_requirements, BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR, CSHARP_CACHE_DIR, - DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM, GO_BIN_CACHE_DIR, GO_CACHE_DIR, - JAVA_CACHE_DIR, NU_CACHE_DIR, POWERSHELL_CACHE_DIR, PY310_CACHE_DIR, PY311_CACHE_DIR, - PY312_CACHE_DIR, PY313_CACHE_DIR, RUBY_CACHE_DIR, RUST_CACHE_DIR, TAR_JAVA_CACHE_DIR, - UV_CACHE_DIR, + get_hub_script_content_and_requirements, init_worker_internal_server_inline_utils, + BUN_BUNDLE_CACHE_DIR, BUN_CACHE_DIR, CSHARP_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, + DENO_CACHE_DIR_NPM, GO_BIN_CACHE_DIR, GO_CACHE_DIR, JAVA_CACHE_DIR, NU_CACHE_DIR, + POWERSHELL_CACHE_DIR, PY310_CACHE_DIR, PY311_CACHE_DIR, PY312_CACHE_DIR, PY313_CACHE_DIR, + RUBY_CACHE_DIR, RUST_CACHE_DIR, TAR_JAVA_CACHE_DIR, UV_CACHE_DIR, }; use crate::monitor::{ @@ -750,9 +750,16 @@ Windmill Community Edition {GIT_VERSION} #[cfg(not(all(feature = "tantivy", feature = "parquet")))] let log_indexer_f = async { Ok(()) as anyhow::Result<()> }; + let worker_internal_server_killpill_rx = killpill_rx.resubscribe(); let server_f = async { if !is_agent { if let Some(db) = conn.as_sql() { + if worker_mode { + init_worker_internal_server_inline_utils( + worker_internal_server_killpill_rx, + base_internal_url.clone(), + )?; + } windmill_api::run_server( db.clone(), index_reader, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 876254d23c..d1fe86e2f0 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -8121,6 +8121,29 @@ paths: type: string format: uuid + /w/{workspace}/jobs/run_inline/preview: + post: + summary: run script preview without starting a new job + operationId: runScriptPreviewInline + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: preview + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/PreviewInline" + + responses: + "200": + description: script result + content: + application/json: + schema: {} + /w/{workspace}/jobs/run_wait_result/preview: post: summary: run script preview and wait for result @@ -16859,6 +16882,18 @@ components: required: - args + PreviewInline: + type: object + properties: + content: + type: string + description: The code to run + args: + $ref: "#/components/schemas/ScriptArgs" + language: + $ref: "#/components/schemas/ScriptLang" + required: [content, args, language] + WorkflowTask: type: object properties: diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index dcb2b6cd08..c1aa63482d 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -28,13 +28,14 @@ use tower::ServiceBuilder; #[cfg(all(feature = "enterprise", feature = "smtp"))] use windmill_common::auth::is_super_admin_email; use windmill_common::auth::TOKEN_PREFIX_LEN; +use windmill_common::client::AuthedClient; use windmill_common::db::UserDbWithAuthed; use windmill_common::error::JsonResult; use windmill_common::flow_conversations::add_message_to_conversation_tx; use windmill_common::flow_status::{JobResult, RestartedFrom}; use windmill_common::jobs::{ check_tag_available_for_workspace_internal, format_completed_job_result, format_result, - DynamicInput, JobTriggerKind, ENTRYPOINT_OVERRIDE, + DynamicInput, JobTriggerKind, RunInlinePreviewScriptFnParams, ENTRYPOINT_OVERRIDE, }; use windmill_common::s3_helpers::{upload_artifact_to_store, BundleFormat}; use windmill_common::utils::{RunnableKind, WarnAfterExt}; @@ -42,6 +43,7 @@ use windmill_common::worker::{Connection, CLOUD_HOSTED, TMP_DIR}; use windmill_common::DYNAMIC_INPUT_CACHE; #[cfg(all(feature = "enterprise", feature = "smtp"))] use windmill_common::{email_oss::send_email_html, server::load_smtp_config}; +use windmill_worker::get_worker_internal_server_inline_utils; use windmill_common::variables::get_workspace_key; @@ -232,6 +234,7 @@ pub fn workspaced_service() -> Router { .layer(ce_headers.clone()), ) .route("/run/preview", post(run_preview_script)) + .route("/run_inline/preview", post(run_inline_preview_script)) .route( "/run_wait_result/preview", post(run_wait_result_preview_script), @@ -3555,6 +3558,13 @@ struct Preview { format: Option, } +#[derive(Debug, Deserialize)] +struct PreviewInline { + content: String, + args: Option>>, + language: ScriptLang, +} + #[derive(Deserialize)] pub struct WorkflowTask { pub args: Option>>, @@ -5670,7 +5680,7 @@ pub async fn stream_job( version, run_query, args, - None + None, ) .await? .0 @@ -5939,6 +5949,39 @@ async fn run_preview_script( Ok((StatusCode::CREATED, uuid.to_string())) } +async fn run_inline_preview_script( + authed: ApiAuthed, + Tokened { token }: Tokened, + Extension(db): Extension, + Path(w_id): Path, + Json(preview): Json, +) -> error::Result { + let utils = get_worker_internal_server_inline_utils()?; + let result = utils.run_inline_preview_script.as_ref()(RunInlinePreviewScriptFnParams { + content: preview.content, + args: preview.args, + workspace_id: w_id.clone(), + base_internal_url: utils.base_internal_url.clone(), + killpill_rx: utils.killpill_rx.resubscribe(), + created_by: authed.display_username().to_string(), + permissioned_as: username_to_permissioned_as(&authed.username), + permissioned_as_email: authed.email.clone(), + lang: preview.language, + job_dir: "".to_string(), + worker_name: "".to_string(), + worker_dir: "".to_string(), + client: AuthedClient { + base_internal_url: utils.base_internal_url.clone(), + force_client: None, + token, + workspace: w_id, + }, + conn: windmill_common::worker::Connection::Sql(db), + }) + .await?; + Ok(Json(to_raw_value(&result)).into_response()) +} + async fn run_wait_result_preview_script( authed: ApiAuthed, Extension(db): Extension, diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index c0a52c6677..7ecd063924 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -68,6 +68,7 @@ aws-credential-types.workspace = true aws-smithy-types.workspace = true base64.workspace = true bitflags.workspace = true +once_cell.workspace = true aws-smithy-types-convert = { workspace = true, optional = true } aws-sdk-rds = { workspace = true, optional = true } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index f32b3447c8..692826bae2 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -1,8 +1,9 @@ -use std::collections::HashMap; +use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc}; use bytes::Bytes; use futures_core::Stream; use indexmap::IndexMap; +use once_cell::sync::OnceCell; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use sqlx::types::Json; @@ -17,6 +18,7 @@ pub const EMAIL_ERROR_HANDLER_USER_EMAIL: &str = "email_error_handler@windmill.d use crate::{ apps::AppScriptId, auth::is_super_admin_email, + client::AuthedClient, db::{AuthedRef, UserDbWithAuthed, DB}, error::{self, to_anyhow, Error}, flow_status::{FlowStatus, RestartedFrom}, @@ -850,3 +852,38 @@ pub async fn lock_debounce_key<'c>( .await .map_err(error::Error::from) } + +pub struct RunInlinePreviewScriptFnParams { + pub workspace_id: String, + pub content: String, + pub lang: ScriptLang, + pub args: Option>>, + pub created_by: String, + pub permissioned_as: String, + pub permissioned_as_email: String, + pub base_internal_url: String, + pub worker_name: String, + pub conn: crate::worker::Connection, + pub client: AuthedClient, + pub job_dir: String, + pub worker_dir: String, + pub killpill_rx: tokio::sync::broadcast::Receiver<()>, +} + +#[derive(Clone)] +pub struct WorkerInternalServerInlineUtils { + pub killpill_rx: Arc>, + pub base_internal_url: String, + pub run_inline_preview_script: Arc< + dyn Fn( + RunInlinePreviewScriptFnParams, + ) -> Pin>> + Send>> + + Send + + Sync, + >, +} +// To run a script inline, bypassing the db and job queue, windmill-api uses these functions. +// windmill-worker sets these functions when it starts up. +// The api cannot call the worker functions directly because they are independent crates +pub static WORKER_INTERNAL_SERVER_INLINE_UTILS: OnceCell = + OnceCell::new(); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 9b47aa5113..a288b9eb55 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -14,6 +14,8 @@ use futures::TryFutureExt; use tokio::time::sleep; use tokio::time::timeout; use windmill_common::client::AuthedClient; +use windmill_common::jobs::WorkerInternalServerInlineUtils; +use windmill_common::jobs::WORKER_INTERNAL_SERVER_INLINE_UTILS; use windmill_common::scripts::hash_to_codebase_id; use windmill_common::scripts::is_special_codebase_hash; use windmill_common::utils::report_critical_error; @@ -3893,3 +3895,98 @@ pub fn parse_sig_of_lang( None }) } + +pub fn init_worker_internal_server_inline_utils( + killpill_rx: tokio::sync::broadcast::Receiver<()>, + base_internal_url: String, +) -> windmill_common::error::Result<()> { + let utils = WorkerInternalServerInlineUtils { + base_internal_url, + killpill_rx: Arc::new(killpill_rx), + run_inline_preview_script: Arc::new(|params| { + let job = MiniPulledJob { + workspace_id: params.workspace_id, + id: Uuid::new_v4(), + args: params.args.map(Json), + parent_job: None, + created_by: params.created_by, + scheduled_for: chrono::Utc::now(), + started_at: None, + runnable_path: None, + kind: JobKind::Preview, + runnable_id: None, + canceled_reason: None, + canceled_by: None, + permissioned_as: params.permissioned_as, + permissioned_as_email: params.permissioned_as_email, + flow_status: None, + tag: "inline_preview".to_string(), + script_lang: Some(params.lang), + same_worker: true, + pre_run_error: None, + concurrent_limit: None, + concurrency_time_window_s: None, + flow_innermost_root_job: None, + root_job: None, + timeout: None, + flow_step_id: None, + cache_ttl: None, + priority: None, + preprocessed: None, + script_entrypoint_override: None, + trigger: None, + trigger_kind: None, + visible_to_owner: false, + permissioned_as_end_user_email: None, + }; + Box::pin(async move { + let mut mem_peak: i32 = -1; + let mut canceled_by: Option = None; + let mut column_order: Option> = None; + let mut new_args: Option>> = None; + let mut occupancy_metrics = OccupancyMetrics::new(Instant::now()); + let mut has_stream: bool = false; + let mut killpill_rx = params.killpill_rx; + + handle_code_execution_job( + &job, + Some(Arc::new(ScriptData { code: params.content, lock: None })), + ¶ms.conn, + ¶ms.client, + None, + ¶ms.job_dir, + ¶ms.worker_dir, + &mut mem_peak, + &mut canceled_by, + ¶ms.base_internal_url, + ¶ms.worker_name, + &mut column_order, + &mut new_args, + &mut occupancy_metrics, + &mut killpill_rx, + None, + &mut has_stream, + ) + .await + }) + }), + }; + WORKER_INTERNAL_SERVER_INLINE_UTILS + .set(utils) + .map_err(|_| { + error::Error::InternalErr( + "Couldn't set WorkerInternalServerInlineUtils OnceCell".to_string(), + ) + })?; + Ok(()) +} + +pub fn get_worker_internal_server_inline_utils( +) -> windmill_common::error::Result<&'static WorkerInternalServerInlineUtils> { + match WORKER_INTERNAL_SERVER_INLINE_UTILS.get() { + Some(utils) => Ok(utils), + None => Err(error::Error::internal_err( + "worker inline functions are meant to be called from a worker's internal server", + )), + } +} diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 379ac0d17c..bf50c8eb61 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1241,13 +1241,24 @@ export function datatable(name: string = "main"): DataTableSqlTemplateFunction { queryStr += strings[i]; if (i !== strings.length - 1) queryStr += `$${i + 1}`; } - const args = Object.fromEntries(values.map((v, i) => [`arg${i + 1}`, v])); + const args = { + ...Object.fromEntries(values.map((v, i) => [`arg${i + 1}`, v])), + database: `datatable://${name}`, + }; return { queryStr, args, query: async () => { - return ["results from datatable (TODO)"]; + let result = await JobService.runScriptPreviewInline({ + workspace: getWorkspace(), + requestBody: { + args, + content: queryStr, + language: "postgresql", + }, + }); + return result; }, }; };