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

inline run works

Browse Source
This commit is contained in:
Diego Imbert
2025-11-25 16:26:22 +01:00
parent aabd194207
commit bc04f80d19
8 changed files with 242 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+1
View File
@@ -15483,6 +15483,7 @@ dependencies = [
"magic-crypt",
"mail-send",
"object_store",
"once_cell",
"openidconnect",
"opentelemetry",
"opentelemetry-appender-tracing",
backend/src/main.rs
+12 -5
View File
@@ -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,
backend/windmill-api/openapi.yaml
+35
View File
@@ -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:
backend/windmill-api/src/jobs.rs
+45 -2
View File
@@ -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<String>,
}
#[derive(Debug, Deserialize)]
struct PreviewInline {
content: String,
args: Option<HashMap<String, Box<JsonRawValue>>>,
language: ScriptLang,
}
#[derive(Deserialize)]
pub struct WorkflowTask {
pub args: Option<HashMap<String, Box<JsonRawValue>>>,
@@ -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<DB>,
Path(w_id): Path<String>,
Json(preview): Json<PreviewInline>,
) -> error::Result<Response> {
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<DB>,
backend/windmill-common/Cargo.toml
+1
View File
@@ -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 }
backend/windmill-common/src/jobs.rs
+38 -1
View File
@@ -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<HashMap<String, Box<RawValue>>>,
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<tokio::sync::broadcast::Receiver<()>>,
pub base_internal_url: String,
pub run_inline_preview_script: Arc<
dyn Fn(
RunInlinePreviewScriptFnParams,
) -> Pin<Box<dyn Future<Output = error::Result<Box<RawValue>>> + 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<WorkerInternalServerInlineUtils> =
OnceCell::new();
backend/windmill-worker/src/worker.rs
+97
View File
@@ -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<CanceledBy> = None;
let mut column_order: Option<Vec<String>> = None;
let mut new_args: Option<HashMap<String, Box<RawValue>>> = 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 })),
&params.conn,
&params.client,
None,
&params.job_dir,
&params.worker_dir,
&mut mem_peak,
&mut canceled_by,
&params.base_internal_url,
&params.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",
)),
}
}
typescript-client/client.ts
+13 -2
View File
@@ -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;
},
};
};