From d0b24d07d21c2dedbd33c533e9eeaceb41fe6dfe Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 3 Oct 2024 13:11:58 +0200 Subject: [PATCH] deno_core is an optional feature flag (#4473) * full * all * add deno_core as features * all * remove warnings * all --- .github/DockerfileBackendTests | 2 + .github/workflows/backend-test.yml | 4 +- .github/workflows/build-staging-image.yml | 2 +- .github/workflows/docker-image-rpi4.yml | 2 +- .github/workflows/docker-image.yml | 12 +- backend/Cargo.lock | 1 + backend/Cargo.toml | 4 +- backend/src/main.rs | 1 + backend/windmill-worker/Cargo.toml | 37 ++- backend/windmill-worker/build.rs | 16 + backend/windmill-worker/src/bun_executor.rs | 93 +++--- backend/windmill-worker/src/job_logger.rs | 2 +- backend/windmill-worker/src/js_eval.rs | 307 ++++++++++++-------- cli/instance.ts | 39 ++- 14 files changed, 326 insertions(+), 196 deletions(-) diff --git a/.github/DockerfileBackendTests b/.github/DockerfileBackendTests index 9fa9d788ff..03f51de332 100644 --- a/.github/DockerfileBackendTests +++ b/.github/DockerfileBackendTests @@ -43,6 +43,8 @@ RUN wget https://golang.org/dl/go1.21.5.linux-amd64.tar.gz && tar -C /usr/local ENV PATH="${PATH}:/usr/local/go/bin" ENV GO_PATH=/usr/local/go/bin/go +RUN curl -LsSf https://astral.sh/uv/install.sh | sh + ENV TZ=Etc/UTC ENV PYTHON_VERSION 3.11.4 diff --git a/.github/workflows/backend-test.yml b/.github/workflows/backend-test.yml index 9963ecd4d1..bedd324f9e 100644 --- a/.github/workflows/backend-test.yml +++ b/.github/workflows/backend-test.yml @@ -44,5 +44,5 @@ jobs: mkdir frontend/build && cd backend && touch windmill-api/openapi-deref.yaml && DATABASE_URL=postgres://postgres:changeme@postgres:5432/windmill - DISABLE_EMBEDDING=true RUST_LOG=info cargo test --features enterprise - --all -- --nocapture + DISABLE_EMBEDDING=true RUST_LOG=info cargo test --features + enterprise,deno_core --all -- --nocapture diff --git a/.github/workflows/build-staging-image.yml b/.github/workflows/build-staging-image.yml index 744ae83aae..724fde4e59 100644 --- a/.github/workflows/build-staging-image.yml +++ b/.github/workflows/build-staging-image.yml @@ -62,7 +62,7 @@ jobs: platforms: linux/amd64,linux/arm64 push: true build-args: | - features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc + features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,deno_core tags: | ${{ steps.meta-ee-public.outputs.tags }} labels: | diff --git a/.github/workflows/docker-image-rpi4.yml b/.github/workflows/docker-image-rpi4.yml index 9fc20c3a87..71a796f679 100644 --- a/.github/workflows/docker-image-rpi4.yml +++ b/.github/workflows/docker-image-rpi4.yml @@ -67,7 +67,7 @@ jobs: platforms: linux/amd64,linux/arm64 push: true build-args: | - features=embedding,parquet,openidconnect + features=embedding,parquet,openidconnect,deno_core tags: | ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:dev ${{ steps.meta-public.outputs.tags }} diff --git a/.github/workflows/docker-image.yml b/.github/workflows/docker-image.yml index 8cac56c35b..7089a870c9 100644 --- a/.github/workflows/docker-image.yml +++ b/.github/workflows/docker-image.yml @@ -1,8 +1,10 @@ env: REGISTRY: ghcr.io - IMAGE_NAME: ${{ github.event_name != 'pull_request' && github.repository || + IMAGE_NAME: + ${{ github.event_name != 'pull_request' && github.repository || 'windmill-labs/windmill-test' }} - DEV_SHA: ${{ github.event_name != 'pull_request' && 'dev' || format('pr-{0}', + DEV_SHA: + ${{ github.event_name != 'pull_request' && 'dev' || format('pr-{0}', github.event.number) }} name: Build windmill:main @@ -75,7 +77,7 @@ jobs: platforms: linux/amd64,linux/arm64 push: true build-args: | - features=embedding,parquet,openidconnect,jemalloc + features=embedding,parquet,openidconnect,jemalloc,deno_core tags: | ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ env.DEV_SHA }} ${{ steps.meta-public.outputs.tags }} @@ -136,7 +138,7 @@ jobs: platforms: linux/amd64,linux/arm64 push: true build-args: | - features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy + features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core tags: | ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }} ${{ steps.meta-ee-public.outputs.tags }} @@ -198,7 +200,7 @@ jobs: platforms: linux/amd64 push: true build-args: | - features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy + features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core PYTHON_IMAGE=python:3.12.2-slim-bookworm tags: | ${{ steps.meta-ee-public-py312.outputs.tags }} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index bbd1c7cdb6..39f4a3e7d4 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -10919,6 +10919,7 @@ dependencies = [ "serde_json", "sha2 0.10.8", "sqlx", + "swc_ecma_parser 0.144.3", "tar", "tiberius", "tokio", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 692edd8f0b..1537837c3f 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -57,6 +57,7 @@ cloud = ["windmill-queue/cloud", "windmill-worker/cloud"] jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"] tantivy = ["dep:windmill-indexer", "windmill-api/tantivy"] sqlx = ["windmill-worker/sqlx"] +deno_core = ["windmill-worker/deno_core", "dep:deno_core"] [dependencies] anyhow.workspace = true @@ -85,7 +86,7 @@ uuid.workspace = true gethostname.workspace = true serde_json.workspace = true serde.workspace = true -deno_core.workspace = true +deno_core = { workspace = true, optional = true } object_store = { workspace = true, optional = true } pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true, default-features = false, features = ['rt_tokio']} quote.workspace = true @@ -105,6 +106,7 @@ serde.workspace = true windmill-api-client.workspace = true deno_core = { workspace = true, features = ["include_js_files_for_snapshotting", "unsafe_use_unprotected_platform"] } + [workspace.dependencies] windmill-api = { path = "./windmill-api", default-features = false } windmill-queue = { path = "./windmill-queue" } diff --git a/backend/src/main.rs b/backend/src/main.rs index 0a59e4af07..3d97e5f1f7 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -118,6 +118,7 @@ where } pub fn main() -> anyhow::Result<()> { + #[cfg(feature = "deno_core")] deno_core::JsRuntime::init_platform(None); create_and_run_current_thread_inner(windmill_main()) } diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index c05bc26fc3..e5b47b7a81 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -18,6 +18,7 @@ parquet = ["windmill-common/parquet", "dep:object_store"] flow_testing = [] cloud = [] sqlx = [] +deno_core = ["dep:deno_fetch", "dep:deno_webidl", "dep:deno_web", "dep:deno_net", "dep:deno_console", "dep:deno_url", "dep:deno_core", "dep:deno_ast", "dep:deno_tls"] [dependencies] windmill-queue.workspace = true @@ -59,15 +60,15 @@ once_cell.workspace = true rsmq_async.workspace = true tokio-postgres.workspace = true bit-vec.workspace = true -deno_fetch.workspace = true -deno_webidl.workspace = true -deno_web.workspace = true -deno_net.workspace = true -deno_console.workspace = true -deno_url.workspace = true -deno_core.workspace = true -deno_ast.workspace = true -deno_tls.workspace = true +deno_fetch = { workspace = true, optional = true } +deno_webidl = { workspace = true, optional = true } +deno_web = { workspace = true, optional = true } +deno_net = { workspace = true, optional = true } +deno_console = { workspace = true, optional = true } +deno_url = { workspace = true, optional = true } +deno_core = { workspace = true, optional = true } +deno_ast = { workspace = true, optional = true } +deno_tls = { workspace = true, optional = true } postgres-native-tls.workspace = true native-tls.workspace = true mysql_async.workspace = true @@ -89,13 +90,17 @@ tar.workspace = true object_store = { workspace = true, optional = true} convert_case.workspace = true yaml-rust.workspace = true +swc_ecma_parser.workspace = true + [build-dependencies] -deno_fetch.workspace = true -deno_webidl.workspace = true -deno_web.workspace = true -deno_console.workspace = true -deno_url.workspace = true -deno_core.workspace = true -deno_net.workspace = true +deno_fetch = { workspace = true, optional = true } +deno_webidl = { workspace = true, optional = true } +deno_web = { workspace = true, optional = true } +deno_net = { workspace = true, optional = true } +deno_console = { workspace = true, optional = true } +deno_url = { workspace = true, optional = true } +deno_core = { workspace = true, optional = true } +deno_ast = { workspace = true, optional = true } +deno_tls = { workspace = true, optional = true } zstd.workspace = true diff --git a/backend/windmill-worker/build.rs b/backend/windmill-worker/build.rs index 8fa581fd5c..66bc66959f 100644 --- a/backend/windmill-worker/build.rs +++ b/backend/windmill-worker/build.rs @@ -1,13 +1,22 @@ +#[cfg(feature = "deno_core")] use deno_fetch::FetchPermissions; +#[cfg(feature = "deno_core")] use deno_net::NetPermissions; +#[cfg(feature = "deno_core")] use deno_web::{BlobStore, TimersPermission}; +#[cfg(feature = "deno_core")] use std::env; +#[cfg(feature = "deno_core")] use std::io::Write; +#[cfg(feature = "deno_core")] use std::path::PathBuf; +#[cfg(feature = "deno_core")] use std::sync::Arc; +#[cfg(feature = "deno_core")] pub struct PermissionsContainer; +#[cfg(feature = "deno_core")] impl FetchPermissions for PermissionsContainer { #[inline(always)] fn check_net_url( @@ -28,6 +37,7 @@ impl FetchPermissions for PermissionsContainer { } } +#[cfg(feature = "deno_core")] impl TimersPermission for PermissionsContainer { #[inline(always)] fn allow_hrtime(&mut self) -> bool { @@ -35,6 +45,7 @@ impl TimersPermission for PermissionsContainer { } } +#[cfg(feature = "deno_core")] impl NetPermissions for PermissionsContainer { fn check_read( &mut self, @@ -61,12 +72,14 @@ impl NetPermissions for PermissionsContainer { } } +#[cfg(feature = "deno_core")] deno_core::extension!( fetch, esm_entry_point = "ext:fetch/src/runtime.js", esm = ["src/runtime.js"], ); +#[cfg(feature = "deno_core")] fn main() { println!("cargo:rustc-env=TARGET={}", env::var("TARGET").unwrap()); println!("cargo:rustc-env=PROFILE={}", env::var("PROFILE").unwrap()); @@ -122,3 +135,6 @@ fn main() { println!("cargo:rerun-if-changed={}", path.display()); } } + +#[cfg(not(feature = "deno_core"))] +fn main() {} diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index e17f140b32..b63ae6c001 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1,8 +1,14 @@ -use std::{collections::HashMap, fs, io, path::Path, process::Stdio, time::Instant}; +#[cfg(feature = "deno_core")] +use std::time::Instant; +use std::{collections::HashMap, fs, io, path::Path, process::Stdio}; use base64::Engine; use itertools::Itertools; + +#[cfg(not(feature = "deno_core"))] +use serde_json::value::to_raw_value; use serde_json::value::RawValue; + use sha2::Digest; use uuid::Uuid; use windmill_parser_ts::remove_pinned_imports; @@ -1190,51 +1196,58 @@ try {{ } } if annotation.native_mode { - let env_code = format!( + #[cfg(not(feature = "deno_core"))] + return Ok(to_raw_value("").unwrap()); + + #[cfg(feature = "deno_core")] + { + let env_code = format!( "const process = {{ env: {{}} }};\nconst BASE_URL = '{base_internal_url}';\nconst BASE_INTERNAL_URL = '{base_internal_url}';\nprocess.env['BASE_URL'] = BASE_URL;process.env['BASE_INTERNAL_URL'] = BASE_INTERNAL_URL;\n{}", reserved_variables .iter() .map(|(k, v)| format!("process.env['{}'] = '{}';\n", k, v)) .collect::>() .join("\n")); - let js_code = read_file_content(&format!("{job_dir}/main.js")).await?; - let started_at = Instant::now(); - let args = crate::common::build_args_map(job, client, db) - .await? - .map(sqlx::types::Json); - let job_args = if args.is_some() { - args.as_ref() - } else { - job.args.as_ref() - }; - let result = crate::js_eval::eval_fetch_timeout( - env_code, - inner_content.clone(), - js_code, - job_args, - job.id, - job.timeout, - db, - mem_peak, - canceled_by, - worker_name, - &job.workspace_id, - false, - occupancy_metrics, - ) - .await?; - tracing::info!( - "Executed native code in {}ms", - started_at.elapsed().as_millis() - ); - append_logs( - &job.id, - &job.workspace_id, - format!("{}\n{}", init_logs, result.1), - db, - ) - .await; - return Ok(result.0); + let js_code = read_file_content(&format!("{job_dir}/main.js")).await?; + let started_at = Instant::now(); + let args = crate::common::build_args_map(job, client, db) + .await? + .map(sqlx::types::Json); + let job_args = if args.is_some() { + args.as_ref() + } else { + job.args.as_ref() + }; + + let result = crate::js_eval::eval_fetch_timeout( + env_code, + inner_content.clone(), + js_code, + job_args, + job.id, + job.timeout, + db, + mem_peak, + canceled_by, + worker_name, + &job.workspace_id, + false, + occupancy_metrics, + ) + .await?; + tracing::info!( + "Executed native code in {}ms", + started_at.elapsed().as_millis() + ); + append_logs( + &job.id, + &job.workspace_id, + format!("{}\n{}", init_logs, result.1), + db, + ) + .await; + return Ok(result.0); + } } append_logs(&job.id, &job.workspace_id, init_logs, db).await; diff --git a/backend/windmill-worker/src/job_logger.rs b/backend/windmill-worker/src/job_logger.rs index 5193fe79aa..05b9e5812b 100644 --- a/backend/windmill-worker/src/job_logger.rs +++ b/backend/windmill-worker/src/job_logger.rs @@ -1,5 +1,5 @@ -use deno_ast::swc::parser::lexer::util::CharExt; use itertools::Itertools; +use swc_ecma_parser::lexer::util::CharExt; #[cfg(all(feature = "enterprise", feature = "parquet"))] use object_store::path::Path; diff --git a/backend/windmill-worker/src/js_eval.rs b/backend/windmill-worker/src/js_eval.rs index 24f83c86af..23029202e5 100644 --- a/backend/windmill-worker/src/js_eval.rs +++ b/backend/windmill-worker/src/js_eval.rs @@ -6,56 +6,72 @@ * LICENSE-AGPL for a copy of the license. */ +#[cfg(feature = "deno_core")] use std::{ cell::RefCell, - collections::HashMap, env, io::{self, BufReader}, rc::Rc, - sync::Arc, }; +use std::{collections::HashMap, sync::Arc}; + +#[cfg(feature = "deno_core")] use deno_ast::ParseParams; +#[cfg(feature = "deno_core")] use deno_core::{ error::AnyError, op2, serde_v8, url, v8::{self, IsolateHandle}, Extension, JsRuntime, OpState, PollEventLoopOptions, RuntimeOptions, }; +#[cfg(feature = "deno_core")] use deno_fetch::FetchPermissions; +#[cfg(feature = "deno_core")] use deno_net::NetPermissions; +#[cfg(feature = "deno_core")] use deno_tls::{rustls::RootCertStore, rustls_pemfile}; +#[cfg(feature = "deno_core")] use deno_web::{BlobStore, TimersPermission}; +#[cfg(feature = "deno_core")] use itertools::Itertools; use lazy_static::lazy_static; use regex::Regex; use serde_json::value::RawValue; use sqlx::types::Json; + +#[cfg(feature = "deno_core")] use tokio::{ sync::{mpsc, oneshot}, time::timeout, }; use uuid::Uuid; -use windmill_common::{error::Error, flow_status::JobResult, DB}; + +#[cfg(feature = "deno_core")] +use windmill_common::error::Error; + +use windmill_common::{flow_status::JobResult, DB}; use windmill_queue::CanceledBy; -use crate::{ - common::{unsafe_raw, OccupancyMetrics}, - handle_child::run_future_with_polling_update_job_poller, - AuthedClient, -}; +use crate::{common::OccupancyMetrics, AuthedClient}; + +#[cfg(feature = "deno_core")] +use crate::{common::unsafe_raw, handle_child::run_future_with_polling_update_job_poller}; #[derive(Debug, Clone)] pub struct IdContext { pub flow_job: Uuid, + #[allow(dead_code)] pub steps_results: HashMap, pub previous_id: String, } +#[cfg(feature = "deno_core")] pub struct ContainerRootCertStoreProvider { root_cert_store: RootCertStore, } +#[cfg(feature = "deno_core")] impl ContainerRootCertStoreProvider { fn new() -> ContainerRootCertStoreProvider { return ContainerRootCertStoreProvider { @@ -73,14 +89,17 @@ impl ContainerRootCertStoreProvider { } } +#[cfg(feature = "deno_core")] impl deno_tls::RootCertStoreProvider for ContainerRootCertStoreProvider { fn get_or_try_init(&self) -> Result<&RootCertStore, AnyError> { Ok(&self.root_cert_store) } } +#[cfg(feature = "deno_core")] pub struct PermissionsContainer; +#[cfg(feature = "deno_core")] impl FetchPermissions for PermissionsContainer { #[inline(always)] fn check_net_url( @@ -101,6 +120,7 @@ impl FetchPermissions for PermissionsContainer { } } +#[cfg(feature = "deno_core")] impl TimersPermission for PermissionsContainer { #[inline(always)] fn allow_hrtime(&mut self) -> bool { @@ -108,6 +128,7 @@ impl TimersPermission for PermissionsContainer { } } +#[cfg(feature = "deno_core")] impl NetPermissions for PermissionsContainer { fn check_read( &mut self, @@ -134,6 +155,7 @@ impl NetPermissions for PermissionsContainer { } } +#[cfg(feature = "deno_core")] pub struct OptAuthedClient(Option); pub async fn eval_timeout( @@ -142,7 +164,7 @@ pub async fn eval_timeout( flow_input: Option>>>, authed_client: Option<&AuthedClient>, by_id: Option, - ctx: Option>, + #[allow(unused_variables)] ctx: Option>, ) -> anyhow::Result> { let expr = expr.trim().to_string(); @@ -212,121 +234,133 @@ pub async fn eval_timeout( } } - let expr2 = expr.clone(); - let (sender, mut receiver) = oneshot::channel::(); - let has_client = authed_client.is_some(); - let authed_client = authed_client.cloned(); - timeout( - std::time::Duration::from_millis(10000), - tokio::task::spawn_blocking(move || { - let mut ops = vec![op_get_context()]; + #[cfg(not(feature = "deno_core"))] + { + #[allow(unreachable_code)] + return todo!(); + } - if authed_client.is_some() { - ops.extend([ - // An op for summing an array of numbers - // The op-layer automatically deserializes inputs - // and serializes the returned Result & value - op_variable(), - op_resource(), - ]) - } + #[cfg(feature = "deno_core")] + { + let expr2 = expr.clone(); + let (sender, mut receiver) = oneshot::channel::(); + let has_client = authed_client.is_some(); + let authed_client = authed_client.cloned(); + return timeout( + std::time::Duration::from_millis(10000), + tokio::task::spawn_blocking(move || { + let mut ops = vec![op_get_context()]; - if by_id.is_some() && authed_client.is_some() { - ops.push(op_get_result()); - ops.push(op_get_id()); - } - - let ext = Extension { name: "js_eval", ops: ops.into(), ..Default::default() }; - let exts = vec![ext]; - // Use our snapshot to provision our new runtime - let options = RuntimeOptions { - extensions: exts, - // startup_snapshot: Some(Snapshot::Static(buffer)), - ..Default::default() - }; - - let mut context_keys = transform_context - .keys() - .filter(|x| expr.contains(&x.to_string())) - .map(|x| x.clone()) - .collect_vec(); - - if !context_keys.contains(&"previous_result".to_string()) - && (p_ids.is_some() && p_ids.as_ref().unwrap().iter().any(|x| expr.contains(x))) - || expr.contains("error") - { - // tracing::error!("PREVIOUS_RESULT"); - context_keys.push("previous_result".to_string()); - } - let has_flow_input = expr.contains("flow_input"); - if has_flow_input { - context_keys.push("flow_input".to_string()) - } - - let mut js_runtime = JsRuntime::new(options); - { - let op_state = js_runtime.op_state(); - let mut op_state = op_state.borrow_mut(); - let mut client = authed_client.clone(); - if let Some(client) = client.as_mut() { - client.force_client = Some( - reqwest::ClientBuilder::new() - .user_agent("windmill/beta") - .danger_accept_invalid_certs( - std::env::var("ACCEPT_INVALID_CERTS").is_ok(), - ) - .build() - .unwrap(), - ); + if authed_client.is_some() { + ops.extend([ + // An op for summing an array of numbers + // The op-layer automatically deserializes inputs + // and serializes the returned Result & value + op_variable(), + op_resource(), + ]) } - op_state.put(OptAuthedClient(client)); - op_state.put(TransformContext { - flow_input: if has_flow_input { flow_input } else { None }, - envs: transform_context - .into_iter() - .filter(|(a, _)| context_keys.contains(a)) - .collect(), - }) - } - sender - .send(js_runtime.v8_isolate().thread_safe_handle()) - .map_err(|_| Error::ExecutionErr("impossible to send v8 isolate".to_string()))?; + if by_id.is_some() && authed_client.is_some() { + ops.push(op_get_result()); + ops.push(op_get_id()); + } - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build()?; + let ext = Extension { name: "js_eval", ops: ops.into(), ..Default::default() }; + let exts = vec![ext]; + // Use our snapshot to provision our new runtime + let options = RuntimeOptions { + extensions: exts, + // startup_snapshot: Some(Snapshot::Static(buffer)), + ..Default::default() + }; - // pretty frail but this it to make the expr more user friendly and not require the user to write await - let expr = ["variable", "resource"] - .into_iter() - .fold(expr, replace_with_await); + let mut context_keys = transform_context + .keys() + .filter(|x| expr.contains(&x.to_string())) + .map(|x| x.clone()) + .collect_vec(); - let expr = replace_with_await_result(expr); + if !context_keys.contains(&"previous_result".to_string()) + && (p_ids.is_some() && p_ids.as_ref().unwrap().iter().any(|x| expr.contains(x))) + || expr.contains("error") + { + // tracing::error!("PREVIOUS_RESULT"); + context_keys.push("previous_result".to_string()); + } + let has_flow_input = expr.contains("flow_input"); + if has_flow_input { + context_keys.push("flow_input".to_string()) + } - let r = runtime.block_on(eval( - &mut js_runtime, - &expr, - context_keys, - by_id, - has_client, - ctx, - ))?; + let mut js_runtime = JsRuntime::new(options); + { + let op_state = js_runtime.op_state(); + let mut op_state = op_state.borrow_mut(); + let mut client = authed_client.clone(); + if let Some(client) = client.as_mut() { + client.force_client = Some( + reqwest::ClientBuilder::new() + .user_agent("windmill/beta") + .danger_accept_invalid_certs( + std::env::var("ACCEPT_INVALID_CERTS").is_ok(), + ) + .build() + .unwrap(), + ); + } + op_state.put(OptAuthedClient(client)); + op_state.put(TransformContext { + flow_input: if has_flow_input { flow_input } else { None }, + envs: transform_context + .into_iter() + .filter(|(a, _)| context_keys.contains(a)) + .collect(), + }) + } - Ok(r) as anyhow::Result> - }), - ) - .await - .map_err(|_| { - if let Ok(isolate) = receiver.try_recv() { - isolate.terminate_execution(); - }; - Error::ExecutionErr(format!( - "The expression of evaluation `{expr2}` took too long to execute (>10000ms)" - )) - })?? + sender + .send(js_runtime.v8_isolate().thread_safe_handle()) + .map_err(|_| { + Error::ExecutionErr("impossible to send v8 isolate".to_string()) + })?; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + + // pretty frail but this it to make the expr more user friendly and not require the user to write await + let expr = ["variable", "resource"] + .into_iter() + .fold(expr, replace_with_await); + + let expr = replace_with_await_result(expr); + + let r = runtime.block_on(eval( + &mut js_runtime, + &expr, + context_keys, + by_id, + has_client, + ctx, + ))?; + + Ok(r) as anyhow::Result> + }), + ) + .await + .map_err(|_| { + if let Ok(isolate) = receiver.try_recv() { + isolate.terminate_execution(); + }; + Error::ExecutionErr(format!( + "The expression of evaluation `{expr2}` took too long to execute (>10000ms)" + )) + })??; + } } +#[cfg(feature = "deno_core")] fn replace_with_await(expr: String, fn_name: &str) -> String { let sep = format!("{}(", fn_name); let mut split = expr.split(&sep); @@ -345,10 +379,12 @@ lazy_static! { Regex::new(r"^(https?)://(([^:@\s]+):([^:@\s]+)@)?([^:@\s]+)(:(\d+))?$").unwrap(); } +#[cfg(feature = "deno_core")] fn replace_with_await_result(expr: String) -> String { RE.replace_all(&expr, "(await $r)").to_string() } +#[cfg(feature = "deno_core")] fn add_closing_bracket(s: &str) -> String { let mut s = s.to_string(); let mut level = 1; @@ -368,6 +404,7 @@ fn add_closing_bracket(s: &str) -> String { s } +#[cfg(feature = "deno_core")] async fn eval( context: &mut JsRuntime, expr: &str, @@ -508,6 +545,7 @@ function get_from_env(name) {{ // } // TODO: Can we a) share the api configuration here somehow or b) just implement this natively in deno, via the deno client? +#[cfg(feature = "deno_core")] #[op2(async)] #[string] async fn op_variable( @@ -522,6 +560,7 @@ async fn op_variable( } } +#[cfg(feature = "deno_core")] #[op2(async)] #[string] async fn op_get_result( @@ -540,6 +579,7 @@ async fn op_get_result( } } +#[cfg(feature = "deno_core")] #[op2(async)] #[string] async fn op_get_id( @@ -563,6 +603,7 @@ async fn op_get_id( } } +#[cfg(feature = "deno_core")] #[op2(async)] #[string] async fn op_resource( @@ -580,11 +621,13 @@ async fn op_resource( } } +#[cfg(feature = "deno_core")] pub struct TransformContext { pub envs: HashMap>>, pub flow_input: Option>>>, } +#[cfg(feature = "deno_core")] #[op2] #[string] fn op_get_context(op_state: Rc>, #[string] id: &str) -> String { @@ -605,6 +648,7 @@ fn op_get_context(op_state: Rc>, #[string] id: &str) -> String } } +#[cfg(feature = "deno_core")] pub fn transpile_ts(expr: String) -> anyhow::Result { let parsed = deno_ast::parse_module(ParseParams { specifier: url::Url::parse("file:///eval.ts")?, @@ -621,21 +665,30 @@ pub fn transpile_ts(expr: String) -> anyhow::Result { .text) } +#[cfg(not(feature = "deno_core"))] +pub fn transpile_ts(_expr: String) -> anyhow::Result { + Ok("require deno".to_string()) +} + +#[cfg(feature = "deno_core")] static RUNTIME_SNAPSHOT: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/FETCH_SNAPSHOT.bin")); +#[cfg(feature = "deno_core")] pub struct MainArgs { args: Vec>>, } +#[cfg(feature = "deno_core")] pub struct LogString { pub s: String, } +#[cfg(feature = "deno_core")] pub struct NativeAnnotation { pub useragent: Option, pub proxy: Option<(String, Option<(String, String)>)>, } - +#[cfg(feature = "deno_core")] pub fn get_annotation(inner_content: &str) -> NativeAnnotation { let mut res = NativeAnnotation { useragent: None, proxy: None }; @@ -655,6 +708,7 @@ pub fn get_annotation(inner_content: &str) -> NativeAnnotation { res } +#[cfg(feature = "deno_core")] fn capture_proxy(s: &str) -> Option<(String, Option<(String, String)>)> { RE_PROXY.captures(s).map(|x| { ( @@ -675,7 +729,27 @@ fn capture_proxy(s: &str) -> Option<(String, Option<(String, String)>)> { ) }) } +#[cfg(not(feature = "deno_core"))] +pub async fn eval_fetch_timeout( + _env_code: String, + _ts_expr: String, + _js_expr: String, + _args: Option<&Json>>>, + _job_id: Uuid, + _job_timeout: Option, + _db: &DB, + _mem_peak: &mut i32, + _canceled_by: &mut Option, + _worker_name: &str, + _w_id: &str, + _load_client: bool, + _occupation_metrics: &mut OccupancyMetrics, +) -> anyhow::Result<(Box, String)> { + use serde_json::value::to_raw_value; + Ok((to_raw_value("require deno_core").unwrap(), "".to_string())) +} +#[cfg(feature = "deno_core")] pub async fn eval_fetch_timeout( env_code: String, ts_expr: String, @@ -851,8 +925,10 @@ pub async fn eval_fetch_timeout( Ok((res, format!("{extra_logs}{logs}"))) } +#[cfg(feature = "deno_core")] const WINDMILL_CLIENT: &str = include_str!("./windmill-client.js"); +#[cfg(feature = "deno_core")] async fn eval_fetch( js_runtime: &mut JsRuntime, expr: &str, @@ -898,6 +974,7 @@ import("file:///eval.ts").then((module) => module.main(...args)).then(JSON.strin Ok(unsafe_raw(r.unwrap_or_else(|| "null".to_string()))) } +#[cfg(feature = "deno_core")] #[op2] #[serde] fn op_get_static_args(op_state: Rc>) -> Vec> { @@ -910,6 +987,7 @@ fn op_get_static_args(op_state: Rc>) -> Vec> { .collect_vec() } +#[cfg(feature = "deno_core")] #[op2(fast)] fn op_log(op_state: Rc>, #[string] log: &str) { // tracing::error!("log: |{}|", log); @@ -920,6 +998,7 @@ fn op_log(op_state: Rc>, #[string] log: &str) { .push_str(log); } +#[cfg(feature = "deno_core")] #[cfg(test)] mod tests { diff --git a/cli/instance.ts b/cli/instance.ts index cf2d5f5b02..2354acdec4 100644 --- a/cli/instance.ts +++ b/cli/instance.ts @@ -183,10 +183,13 @@ export type InstanceSyncOptions = { yes?: boolean; }; -export async function pickInstance(opts: InstanceSyncOptions, allowNew: boolean) { +export async function pickInstance( + opts: InstanceSyncOptions, + allowNew: boolean +) { const instances = await allInstances(); if (opts.baseUrl && opts.token) { - log.info("Using instance fully defined by --base-url and --token") + log.info("Using instance fully defined by --base-url and --token"); return { name: "custom", remote: opts.baseUrl, @@ -335,12 +338,14 @@ async function instancePull(opts: GlobalOptions & InstanceSyncOptions) { ); if (localWorkspacesToDelete.length > 0) { - const confirmDelete = await Confirm.prompt({ - message: - "Do you want to delete the local copy of workspaces that don't exist anymore on the instance?\n" + - localWorkspacesToDelete.map((w) => w.workspaceId).join(", "), - default: true, - }); + const confirmDelete = + opts.yes || + (await Confirm.prompt({ + message: + "Do you want to delete the local copy of workspaces that don't exist anymore on the instance?\n" + + localWorkspacesToDelete.map((w) => w.workspaceId).join(", "), + default: true, + })); if (confirmDelete) { for (const workspace of localWorkspacesToDelete) { @@ -485,12 +490,14 @@ async function instancePush(opts: GlobalOptions & InstanceSyncOptions) { (w) => !localWorkspaces.find((l) => l.workspaceId === w.id) ); if (workspacesToDelete.length > 0) { - const confirmDelete = await Confirm.prompt({ - message: - "Do you want to delete the following remote workspaces that don't exist locally?\n" + - workspacesToDelete.map((w) => w.id).join(", "), - default: true, - }); + const confirmDelete = + opts.yes || + (await Confirm.prompt({ + message: + "Do you want to delete the following remote workspaces that don't exist locally?\n" + + workspacesToDelete.map((w) => w.id).join(", "), + default: true, + })); if (confirmDelete) { for (const workspace of workspacesToDelete) { @@ -544,7 +551,9 @@ async function whoami(opts: {}) { log.info(colors.green.underline(`global whoami infos:`)); log.info(JSON.stringify(whoamiInfo, null, 2)); } catch (error) { - log.error(colors.red(`Failed to retrieve whoami information: ${error.message}`)); + log.error( + colors.red(`Failed to retrieve whoami information: ${error.message}`) + ); } }