diff --git a/Dockerfile b/Dockerfile index e2d5653024..837e355a61 100644 --- a/Dockerfile +++ b/Dockerfile @@ -111,12 +111,10 @@ FROM ${PYTHON_IMAGE} ARG TARGETPLATFORM ARG POWERSHELL_VERSION=7.3.5 ARG POWERSHELL_DEB_VERSION=7.3.5-1 -ARG RCLONE_VERSION=1.60.1 ARG KUBECTL_VERSION=1.27.2 ARG HELM_VERSION=3.12.0 ARG APP=/usr/src/app ARG WITH_POWERSHELL=true -ARG WITH_RCLONE=true ARG WITH_KUBECTL=true ARG WITH_HELM=true @@ -167,12 +165,6 @@ RUN set -eux; \ unzip awscliv2.zip && \ ./aws/install && rm awscliv2.zip -RUN if [ "$WITH_RCLONE" = "true" ]; then \ - arch="$(dpkg --print-architecture)"; arch="${arch##*-}"; \ - curl -o rclone.zip "https://downloads.rclone.org/v${RCLONE_VERSION}/rclone-v${RCLONE_VERSION}-linux-$arch.zip"; \ - unzip -p rclone.zip rclone-v${RCLONE_VERSION}-linux-$arch/rclone > /usr/bin/rclone; rm rclone.zip; \ - chown root:root /usr/bin/rclone; chmod 755 /usr/bin/rclone; \ - else echo 'Building the image without rclone'; fi RUN set -eux; \ diff --git a/backend/.cargo/config.toml b/backend/.cargo/config.toml index 7d27c4a1e1..5095e00c71 100644 --- a/backend/.cargo/config.toml +++ b/backend/.cargo/config.toml @@ -1,6 +1,10 @@ [build] incremental = true +[target.x86_64-unknown-linux-gnu] +linker = "/usr/bin/clang" +rustflags = ["-C", "link-arg=--ld-path=/usr/bin/mold"] + [target.x86_64-apple-darwin] rustflags = [ "-C", "link-arg=-undefined", diff --git a/backend/Cargo.lock b/backend/Cargo.lock index b65e7d459c..a61ad3a560 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -9776,6 +9776,7 @@ dependencies = [ "mysql_async", "native-tls", "nix", + "object_store", "once_cell", "openidconnect", "pem 3.0.3", @@ -9790,6 +9791,7 @@ dependencies = [ "serde_json", "sha2 0.10.8", "sqlx", + "tar", "tiberius", "tokio", "tokio-postgres", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 596e7e8236..d786613e44 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -235,3 +235,4 @@ aws-config = "^1" aws-sdk-sts = "^1" crc = "^3" +tar = "^0" \ No newline at end of file diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 211ad1125a..0b0d9f7ca0 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -aa7768505b23651069ae2b18470edc5c0023c45a \ No newline at end of file +73ebc9cb3de7b027f238d60c8ec19da16ae2dc94 \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index d51e396ff6..49a564cb66 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -24,7 +24,7 @@ use windmill_common::{ JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, S3_CACHE_BUCKET_SETTING, SAML_METADATA_SETTING, + RETENTION_PERIOD_SECS_SETTING, S3_CACHE_CONFIG_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, stats_ee::schedule_stats, @@ -48,7 +48,7 @@ use crate::monitor::{ monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting, reload_extra_pip_index_url_setting, reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting, reload_pip_index_url_setting, - reload_retention_period_setting, reload_s3_cache_bucket_setting, reload_scim_token_setting, + reload_retention_period_setting, reload_s3_cache_setting, reload_scim_token_setting, reload_server_config, reload_worker_config, }; @@ -441,8 +441,9 @@ Windmill Community Edition {GIT_VERSION} JOB_DEFAULT_TIMEOUT_SECS_SETTING => { reload_job_default_timeout_setting(&db).await }, - S3_CACHE_BUCKET_SETTING => { - reload_s3_cache_bucket_setting(&db).await + #[cfg(feature = "parquet")] + S3_CACHE_CONFIG_SETTING => { + reload_s3_cache_setting(&db).await }, SCIM_TOKEN_SETTING => { reload_scim_token_setting(&db).await diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 4866aa0591..703f25ec9f 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -19,27 +19,18 @@ use windmill_api::{ DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN, }; use windmill_common::{ - error, - flow_status::FlowStatusModule, - global_settings::{ + error, flow_status::FlowStatusModule, global_settings::{ BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, S3_CACHE_BUCKET_SETTING, SAML_METADATA_SETTING, + RETENTION_PERIOD_SECS_SETTING, S3_CACHE_CONFIG_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, - }, - jobs::QueuedJob, - oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, - s3_helpers::S3_CACHE_BUCKET, - server::load_server_config, - users::truncate_token, - worker::{ + }, jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, s3_helpers::{build_s3_client_from_settings, S3Settings, S3_CACHE_SETTINGS}, server::load_server_config, users::truncate_token, worker::{ load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG, WORKER_CONFIG, - }, - BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED, + }, BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED }; use windmill_queue::cancel_job; use windmill_worker::{ @@ -128,7 +119,9 @@ pub async fn initial_load( tracing::error!("Error reloading base url: {:?}", e) } - reload_s3_cache_bucket_setting(&db).await; + #[cfg(feature = "parquet")] + reload_s3_cache_setting(&db).await; + if server_mode { reload_server_config(&db).await; reload_retention_period_setting(&db).await; @@ -151,12 +144,7 @@ pub async fn initial_load( } pub async fn load_metrics_enabled(db: &DB) -> error::Result<()> { - let metrics_enabled = sqlx::query_scalar!( - "SELECT value FROM global_settings WHERE name = $1", - EXPOSE_METRICS_SETTING - ) - .fetch_optional(db) - .await; + let metrics_enabled = load_value_from_global_settings(db, EXPOSE_METRICS_SETTING).await; match metrics_enabled { Ok(Some(serde_json::Value::Bool(t))) => METRICS_ENABLED.store(t, Ordering::Relaxed), _ => (), @@ -165,12 +153,8 @@ pub async fn load_metrics_enabled(db: &DB) -> error::Result<()> { } pub async fn load_tag_per_workspace_enabled(db: &DB) -> error::Result<()> { - let metrics_enabled = sqlx::query_scalar!( - "SELECT value FROM global_settings WHERE name = $1", - DEFAULT_TAGS_PER_WORKSPACE_SETTING - ) - .fetch_optional(db) - .await; + let metrics_enabled = load_value_from_global_settings(db, DEFAULT_TAGS_PER_WORKSPACE_SETTING).await; + match metrics_enabled { Ok(Some(serde_json::Value::Bool(t))) => { DEFAULT_TAGS_PER_WORKSPACE.store(t, Ordering::Relaxed) @@ -181,11 +165,9 @@ pub async fn load_tag_per_workspace_enabled(db: &DB) -> error::Result<()> { } pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> { - let metrics_enabled = sqlx::query_scalar!( - "SELECT value FROM global_settings WHERE name = $1", + let metrics_enabled = load_value_from_global_settings(db, EXPOSE_DEBUG_METRICS_SETTING ) - .fetch_optional(db) .await; match metrics_enabled { Ok(Some(serde_json::Value::Bool(t))) => METRICS_DEBUG_ENABLED.store(t, Ordering::Relaxed), @@ -195,11 +177,9 @@ pub async fn load_metrics_debug_enabled(db: &DB) -> error::Result<()> { } pub async fn load_keep_job_dir(db: &DB) { - let value = sqlx::query_scalar!( - "SELECT value FROM global_settings WHERE name = $1", + let value = load_value_from_global_settings(db, KEEP_JOB_DIR_SETTING ) - .fetch_optional(db) .await; match value { Ok(Some(serde_json::Value::Bool(t))) => KEEP_JOB_DIR.store(t, Ordering::Relaxed), @@ -211,12 +191,9 @@ pub async fn load_keep_job_dir(db: &DB) { } pub async fn load_require_preexisting_user(db: &DB) { - let value = sqlx::query_scalar!( - "SELECT value FROM global_settings WHERE name = $1", + let value = load_value_from_global_settings(db, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING - ) - .fetch_optional(db) - .await; + ).await; match value { Ok(Some(serde_json::Value::Bool(t))) => { REQUIRE_PREEXISTING_USER_FOR_OAUTH.store(t, Ordering::Relaxed) @@ -411,14 +388,43 @@ pub async fn reload_retention_period_setting(db: &DB) { } } -pub async fn reload_s3_cache_bucket_setting(db: &DB) { - reload_option_setting_with_tracing( - db, - S3_CACHE_BUCKET_SETTING, - "S3_CACHE_BUCKET", - S3_CACHE_BUCKET.clone(), - ) - .await; + +#[cfg(feature = "parquet")] +pub async fn reload_s3_cache_setting(db: &DB) { + let s3_config = load_value_from_global_settings(db, S3_CACHE_CONFIG_SETTING).await; + if let Err(e) = s3_config { + tracing::error!("Error reloading s3 cache config: {:?}", e) + } else { + if let Some(v) = s3_config.unwrap() { + let mut s3_cache_settings = S3_CACHE_SETTINGS.write().await; + let setting = serde_json::from_value::(v); + if let Err(e) = setting { + tracing::error!("Error parsing s3 cache config: {:?}", e) + } else { + let s3_client = build_s3_client_from_settings(setting.unwrap()).await; + if let Err(e) = s3_client { + tracing::error!("Error building s3 client from settings: {:?}", e) + } else { + *s3_cache_settings = Some(s3_client.unwrap()); + } + } + } else { + let mut s3_cache_settings = S3_CACHE_SETTINGS.write().await; + if std::env::var("S3_CACHE_BUCKET").is_ok() { + *s3_cache_settings = build_s3_client_from_settings(S3Settings { + bucket: None, + region: None, + access_key: None, + secret_key: None, + endpoint: None, + store_logs: None, + allow_http: None + }).await.ok(); + } else { + *s3_cache_settings = None; + } + } + } } pub async fn reload_job_default_timeout_setting(db: &DB) { @@ -447,12 +453,9 @@ pub async fn reload_request_size(db: &DB) { } pub async fn reload_license_key(db: &DB) -> error::Result<()> { - let q = sqlx::query!( - "SELECT value FROM global_settings WHERE name = $1", + let q = load_value_from_global_settings(db, LICENSE_KEY_SETTING - ) - .fetch_optional(db) - .await?; + ).await?; let mut value = std::env::var("LICENSE_KEY") .ok() @@ -460,14 +463,14 @@ pub async fn reload_license_key(db: &DB) -> error::Result<()> { .unwrap_or(String::new()); if let Some(q) = q { - if let Ok(v) = serde_json::from_value::(q.value.clone()) { + if let Ok(v) = serde_json::from_value::(q.clone()) { tracing::info!( "Loaded setting LICENSE_KEY from db config: {}", truncate_token(&v) ); value = v; } else { - tracing::error!("Could not parse LICENSE_KEY found: {:#?}", &q.value); + tracing::error!("Could not parse LICENSE_KEY found: {:#?}", &q); } }; @@ -486,32 +489,37 @@ pub async fn reload_option_setting_with_tracing( tracing::error!("Error reloading setting {}: {:?}", setting_name, e) } } + +async fn load_value_from_global_settings(db: &DB, setting_name: &str) -> error::Result> { + let r = sqlx::query!( + "SELECT value FROM global_settings WHERE name = $1", + setting_name + ) + .fetch_optional(db) + .await?.map(|x| x.value); + Ok(r) +} pub async fn reload_option_setting( db: &DB, setting_name: &str, std_env_var: &str, lock: Arc>>, ) -> error::Result<()> { - let q = sqlx::query!( - "SELECT value FROM global_settings WHERE name = $1", - setting_name - ) - .fetch_optional(db) - .await?; + let q = load_value_from_global_settings(db, setting_name).await?; let mut value = std::env::var(std_env_var) .ok() .and_then(|x| x.parse::().ok()); if let Some(q) = q { - if let Ok(v) = serde_json::from_value::(q.value.clone()) { + if let Ok(v) = serde_json::from_value::(q.clone()) { tracing::info!( "Loaded setting {setting_name} from db config: {:#?}", - &q.value + &q ); value = Some(v) } else { - tracing::error!("Could not parse {setting_name} found: {:#?}", &q.value); + tracing::error!("Could not parse {setting_name} found: {:#?}", &q); } }; @@ -534,12 +542,7 @@ pub async fn reload_setting( lock: Arc>, transformer: fn(T) -> T, ) -> error::Result<()> { - let q = sqlx::query!( - "SELECT value FROM global_settings WHERE name = $1", - setting_name - ) - .fetch_optional(db) - .await?; + let q = load_value_from_global_settings(db, setting_name).await?; let mut value = std::env::var(std_env_var) .ok() @@ -547,14 +550,14 @@ pub async fn reload_setting( .unwrap_or(default); if let Some(q) = q { - if let Ok(v) = serde_json::from_value::(q.value.clone()) { + if let Ok(v) = serde_json::from_value::(q.clone()) { tracing::info!( "Loaded setting {setting_name} from db config: {:#?}", - &q.value + &q ); value = transformer(v); } else { - tracing::error!("Could not parse {setting_name} found: {:#?}", &q.value); + tracing::error!("Could not parse {setting_name} found: {:#?}", &q); } }; @@ -727,18 +730,15 @@ pub async fn reload_worker_config( } pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { - let q_base_url = sqlx::query!( - "SELECT value FROM global_settings WHERE name = $1", + let q_base_url = load_value_from_global_settings(db, BASE_URL_SETTING - ) - .fetch_optional(db) - .await?; + ).await?; let std_base_url = std::env::var("BASE_URL") .ok() .unwrap_or_else(|| "http://localhost".to_string()); let base_url = if let Some(q) = q_base_url { - if let Ok(v) = serde_json::from_value::(q.value.clone()) { + if let Ok(v) = serde_json::from_value::(q.clone()) { if v != "" { v } else { @@ -747,7 +747,7 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { } else { tracing::error!( "Could not parse base_url setting as a string, found: {:#?}", - &q.value + &q ); std_base_url } @@ -755,22 +755,20 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> { std_base_url }; - let q_oauth = sqlx::query!( - "SELECT value FROM global_settings WHERE name = $1", + let q_oauth = load_value_from_global_settings(db, OAUTH_SETTING ) - .fetch_optional(db) .await?; let oauths = if let Some(q) = q_oauth { if let Ok(v) = - serde_json::from_value::>>(q.value.clone()) + serde_json::from_value::>>(q.clone()) { v } else { tracing::error!( "Could not parse oauth setting as a json, found: {:#?}", - &q.value + &q ); None } diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a8052d96ae..dae50e579f 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -718,6 +718,38 @@ paths: schema: type: string + /settings/test_s3_config: + post: + summary: test s3 config + operationId: testS3Config + tags: + - setting + requestBody: + description: test s3 config + required: true + content: + application/json: + schema: + type: object + properties: + bucket: + type: string + region: + type: string + access_key: + type: string + secret_key: + type: tring + endpoint: + type: string + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + /settings/send_stats: post: summary: send stats diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs index f6358f06a3..fce1121dc8 100644 --- a/backend/windmill-api/src/settings.rs +++ b/backend/windmill-api/src/settings.rs @@ -39,7 +39,9 @@ pub fn global_service() -> Router { ) .route("/test_smtp", post(test_email)) .route("/test_license_key", post(test_license_key)) + .route("/test_s3_config", post(test_s3_bucket)) .route("/send_stats", post(send_stats)) + } #[derive(Deserialize)] @@ -87,6 +89,38 @@ pub async fn test_email( Ok("Sent test email".to_string()) } +#[cfg(feature = "parquet")] +use windmill_common::s3_helpers::S3Settings; + +#[cfg(feature = "parquet")] +use windmill_common::s3_helpers::build_s3_client_from_settings; + + + +#[cfg(feature = "parquet")] +pub async fn test_s3_bucket( + Extension(db): Extension, + authed: ApiAuthed, + Json(test_s3_bucket): Json, +) -> error::Result { + use bytes::Bytes; + + + + require_super_admin(&db, &authed.email).await?; + let client = build_s3_client_from_settings(test_s3_bucket).await?; + + let path = object_store::path::Path::from(format!("/test-s3-bucket-{uuid}", uuid = uuid::Uuid::new_v4())); + tracing::info!("Testing s3 bucket at path: {path}"); + client.put(&path, Bytes::from_static(b"hello")).await.map_err(to_anyhow)?; + let content = client.get(&path).await.map_err(to_anyhow)?.bytes().await.map_err(to_anyhow)?; + if content != Bytes::from_static(b"hello") { + return Err(error::Error::InternalErr("Failed to read back from s3".to_string())); + } + client.delete(&path).await.map_err(to_anyhow)?; + Ok("Tested bucket successfully".to_string()) +} + #[derive(Deserialize)] pub struct TestKey { pub license_key: String, diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index 0debf05ee7..14be142e57 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -20,7 +20,7 @@ pub const EXPOSE_METRICS_SETTING: &str = "expose_metrics"; pub const EXPOSE_DEBUG_METRICS_SETTING: &str = "expose_debug_metrics"; pub const KEEP_JOB_DIR_SETTING: &str = "keep_job_dir"; pub const REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING: &str = "require_preexisting_user_for_oauth"; -pub const S3_CACHE_BUCKET_SETTING: &str = "s3_cache_bucket"; +pub const S3_CACHE_CONFIG_SETTING: &str = "s3_cache_config"; pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation"; diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index f0fd08874d..1b0faa1214 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -1,5 +1,8 @@ #[cfg(feature = "parquet")] use crate::error; +use aws_sdk_sts::config::ProvideCredentials; +use axum::async_trait; +use object_store::aws::AwsCredential; #[cfg(feature = "parquet")] use object_store::azure::MicrosoftAzureBuilder; #[cfg(feature = "parquet")] @@ -12,8 +15,7 @@ use tokio::sync::RwLock; lazy_static::lazy_static! { - pub static ref S3_CACHE_BUCKET: Arc>> = Arc::new(RwLock::new(None)); - + pub static ref S3_CACHE_SETTINGS: Arc>>> = Arc::new(RwLock::new(None)); } #[derive(Serialize, Deserialize, Debug)] @@ -42,8 +44,8 @@ pub struct AzureBlobStorage { #[derive(Clone, Debug)] pub enum ObjectStoreResource { - S3Resource(S3Resource), - AzureBlobResource(AzureBlobResource), + S3(S3Resource), + Azure(AzureBlobResource), } #[derive(Deserialize, Debug)] @@ -71,6 +73,7 @@ pub struct S3Resource { pub path_style: Option, pub token: Option, pub port: Option, + } #[derive(Debug, Deserialize, Serialize, Clone)] @@ -163,15 +166,20 @@ pub fn build_object_store_client( resource_ref: &ObjectStoreResource, ) -> error::Result> { match resource_ref { - ObjectStoreResource::S3Resource(s3_resource_ref) => build_s3_client(&s3_resource_ref), - ObjectStoreResource::AzureBlobResource(azure_blob_resource_ref) => { + ObjectStoreResource::S3(s3_resource_ref) => build_s3_client(&s3_resource_ref, None), + ObjectStoreResource::Azure(azure_blob_resource_ref) => { build_azure_blob_client(&azure_blob_resource_ref) } } } #[cfg(feature = "parquet")] -fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result> { +use aws_config::{default_provider::credentials::DefaultCredentialsChain, Region}; +#[cfg(feature = "parquet")] +use object_store::CredentialProvider; + +pub fn build_s3_client(s3_resource_ref: &S3Resource, credential_providers: Option) -> error::Result> { + let s3_resource = s3_resource_ref.clone(); let endpoint = render_endpoint( s3_resource.endpoint, @@ -181,12 +189,20 @@ fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result, + pub region: Option, + pub access_key: Option, + pub secret_key: Option, + pub endpoint: Option, + pub allow_http: Option, // default to true + pub store_logs: Option, +} + + fn none_if_empty(s: Option) -> Option { + if s.is_none() || s.as_ref().unwrap().is_empty() { + None + } else { + s + } +} + +#[cfg(feature = "parquet")] +pub async fn build_s3_client_from_settings(settings: S3Settings) -> error::Result> { + let region = none_if_empty(settings.region).unwrap_or_else(|| std::env::var("AWS_REGION").unwrap_or_else(|_| "us-east-1".to_string())); + let access_key = none_if_empty(settings.access_key); + let secret_key = none_if_empty(settings.secret_key); + let credentials_provider = if access_key.is_none() && secret_key.is_none() { + Some(DefaultCredentialsChain::builder() + .region(Region::new(region.clone())) + .build() + .await) + } else { None }; + let s3_resource = S3Resource { + endpoint: none_if_empty(settings.endpoint).unwrap_or_else(|| std::env::var("S3_ENDPOINT") + .unwrap_or_else(|_| format!("s3.{region}.amazonaws.com"))), + bucket: settings.bucket.clone() + .unwrap_or_else(|| std::env::var("S3_CACHE_BUCKET").unwrap_or_else(|_| "missingbucket".to_string())), + region, + access_key, + secret_key, + use_ssl: !settings.allow_http.unwrap_or(false), + path_style: None, + port: None, + token: None, + }; + + + build_s3_client(&s3_resource, credentials_provider) +} + +#[derive(Debug)] +struct AwsCredentialAdapter { + pub inner: DefaultCredentialsChain, +} + +#[async_trait] +impl CredentialProvider for AwsCredentialAdapter { + type Credential = AwsCredential; + async fn get_credential(&self) -> object_store::Result> { + let creds = self.inner.provide_credentials().await.unwrap(); + Ok(Arc::new(Self::Credential { + key_id: creds.access_key_id().to_string(), + secret_key: creds.secret_access_key().to_string(), + token: creds.session_token().map(|s| s.to_string()), + })) + } +} diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index d1baecc2c5..6226c00401 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -14,7 +14,7 @@ prometheus = ["dep:prometheus", "windmill-common/prometheus"] enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util", "dep:openidconnect"] benchmark = ["windmill-queue/benchmark"] flamegraph = [] -parquet = ["windmill-common/parquet"] +parquet = ["windmill-common/parquet", "dep:object_store"] flow_testing = [] [dependencies] @@ -78,6 +78,8 @@ hex.workspace = true tiberius = { workspace = true, optional = true } tokio-util = { workspace = true, optional = true } openidconnect = { workspace = true, optional = true} +tar.workspace = true +object_store = { workspace = true, optional = true} [build-dependencies] deno_fetch.workspace = true @@ -85,4 +87,4 @@ deno_webidl.workspace = true deno_web.workspace = true deno_console.workspace = true deno_url.workspace = true -deno_core.workspace = true \ No newline at end of file +deno_core.workspace = true diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index e98be33166..e8ca2da625 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -638,7 +638,17 @@ plugin(p) .replace("{JOB_DIR}", job_dir) .replace("{CACHE_DIR}", BUN_CACHE_DIR) .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string()) - .replace("{SHARED_MOUNT}", shared_mount), + .replace( + "{SHARED_MOUNT}", + &shared_mount.replace( + "/tmp/shared", + if annotation.nodejs_mode { + "/tmp/nodejs/shared" + } else { + "/tmp/bun/shared" + }, + ), + ), ) .await?; diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index fa424b4651..6b702de39c 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -1,35 +1,34 @@ #[cfg(feature = "enterprise")] use crate::{ROOT_TMP_CACHE_DIR, TAR_PIP_TMP_CACHE_DIR}; -#[cfg(feature = "enterprise")] -use itertools::Itertools; + // #[cfg(feature = "enterprise")] // use rand::Rng; -#[cfg(feature = "enterprise")] -use std::process::Stdio; #[cfg(feature = "enterprise")] -use tokio::{process::Command, time::Instant}; +use tokio::time::Instant; + +#[cfg(feature = "parquet")] +use object_store::ObjectStore; #[cfg(feature = "enterprise")] use windmill_common::error; -#[cfg(feature = "enterprise")] -pub async fn build_tar_and_push(bucket: String, folder: String) -> error::Result<()> { +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use std::sync::Arc; + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +pub async fn build_tar_and_push(s3_client: Arc, folder: String) -> error::Result<()> { + use bytes::Bytes; + use object_store::path::Path; + tracing::info!("Started building and pushing piptar {folder}"); let start = Instant::now(); let folder_name = folder.split("/").last().unwrap(); let tar_path = format!("{TAR_PIP_TMP_CACHE_DIR}/{folder_name}.tar",); - if let Err(e) = execute_command( - ROOT_TMP_CACHE_DIR, - "tar", - vec!["-c", "-f", &tar_path, "-C", &folder, "."], - ) - .await - { - tracing::info!("Failed to tar cache. Error: {:?}", e); - return Err(e); - } + let tar_file = std::fs::File::create(&tar_path)?; + let mut tar = tar::Builder::new(tar_file); + tar.append_dir_all(".", &folder)?; let tar_metadata = tokio::fs::metadata(&tar_path).await; if tar_metadata.is_err() || tar_metadata.as_ref().unwrap().len() == 0 { @@ -39,37 +38,42 @@ pub async fn build_tar_and_push(bucket: String, folder: String) -> error::Result ))); } - let bucket = bucket.trim_start_matches("s3://"); - if let Err(e) = execute_command( - ROOT_TMP_CACHE_DIR, - "rclone", - vec![ - "copyto", - &tar_path, - &format!(":s3,env_auth=true:{bucket}/tar/pip/{folder_name}.tar"), - "-v", - "--size-only", - "--fast-list", - "--s3-no-check-bucket", - ], - ) - .await + // let s3_settings = S3_CACHE_SETTINGS.read().await; + // let s3_client = s3_settings.as_ref().ok_or_else(|| { + // error::Error::ExecutionErr("Failed to read s3 cache settings".to_string()) + // })?; + if let Err(e) = s3_client + .put( + &Path::from(format!("/tar/pip/{folder_name}.tar")), + Bytes::from(std::fs::read(&tar_path)?), + ) + .await { - tracing::info!("Failed to copy piptar {folder} to bucket. Error: {:?}", e); - return Err(e); + tracing::info!("Failed to put tar to s3: {tar_path}. Error: {:?}", e); + return Err(error::Error::ExecutionErr(format!( + "Failed to put tar to s3: {tar_path}" + ))); } + tokio::fs::remove_file(&tar_path).await.map_err(|e| { + tracing::error!("Failed to remove piptar {folder_name}. Error: {:?}", e); + e + })?; + tracing::info!( - "Finished copying piptar {folder} to bucket {bucket} as tar, took: {:?}s. Size of tar: {}", + "Finished copying piptar {folder} to bucket as tar, took: {:?}s. Size of tar: {}", start.elapsed().as_secs(), - tar_metadata.unwrap().len() + tar_metadata.unwrap().len(), ); Ok(()) } -#[cfg(feature = "enterprise")] -pub async fn pull_from_tar(bucket: &str, folder: String) -> error::Result<()> { + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +pub async fn pull_from_tar(client: Arc, folder: String) -> error::Result<()> { + + use object_store::path::Path; use tokio::fs::metadata; let folder_name = folder.split("/").last().unwrap(); @@ -78,29 +82,29 @@ pub async fn pull_from_tar(bucket: &str, folder: String) -> error::Result<()> { let start = Instant::now(); let tar_path = format!("tar/pip/{folder_name}.tar"); let target = format!("{ROOT_TMP_CACHE_DIR}/{tar_path}.single"); - let bucket = bucket.trim_start_matches("s3://"); - if let Err(e) = execute_command( - ROOT_TMP_CACHE_DIR, - "rclone", - vec![ - "copyto", - &format!(":s3,env_auth=true:{bucket}/{tar_path}"), - &target, - "-v", - "--size-only", - "--fast-list", - ], - ) - .await - { - tracing::info!( - "Failed to copy tar {folder_name} from bucket. Error: {:?}", - e - ); - return Err(e); + let object = client + .get(&Path::from(format!("tar/pip/{folder_name}.tar"))) + .await; + if let Err(e) = object { + tracing::info!("Failed to put tar to s3: {tar_path}. Error: {:?}", e); + return Err(error::Error::ExecutionErr(format!( + "Failed to put tar to s3: {tar_path}" + ))); } + use tokio::io::AsyncWriteExt; + let mut file = tokio::fs::OpenOptions::new() + .write(true) + .create(true) + .open(&target) + .await + .unwrap(); + + file.write_all(&object.unwrap().bytes().await.unwrap()) + .await + .unwrap(); + if metadata(&target).await.is_err() { tracing::info!( "piptar {folder_name} not found in bucket. Took {:?}ms", @@ -112,8 +116,15 @@ pub async fn pull_from_tar(bucket: &str, folder: String) -> error::Result<()> { } // tracing::info!("B: {target} {folder}"); - extract_pip_tar(&target, &folder).await?; - tokio::fs::remove_file(&target).await?; + extract_pip_tar(&target, &folder).await.map_err(|e| { + tracing::error!("Failed to extract piptar {folder_name}. Error: {:?}", e); + e + })?; + + tokio::fs::remove_file(&target).await.map_err(|e| { + tracing::error!("Failed to remove piptar {folder_name}. Error: {:?}", e); + e + })?; tracing::info!( "Finished pulling and extracting {folder_name}. Took {:?}ms", start.elapsed().as_millis() @@ -124,14 +135,22 @@ pub async fn pull_from_tar(bucket: &str, folder: String) -> error::Result<()> { #[cfg(feature = "enterprise")] pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> { - use tokio::fs; + use tokio::fs::{self}; let start: Instant = Instant::now(); fs::create_dir_all(&folder).await?; - if let Err(e) = execute_command(&folder, "tar", vec!["-xpvf", tar]).await { + + let mut ar = tar::Archive::new( + std::fs::File::open(tar) + .map_err(|e| error::Error::ExecutionErr(format!("Failed to open tar {tar}: {e}")))?, + ); + + if let Err(e) = ar.unpack(folder) { tracing::info!("Failed to untar piptar. Error: {:?}", e); fs::remove_dir_all(&folder).await?; - return Err(e); + return Err(error::Error::ExecutionErr(format!( + "Failed to untar piptar {folder}" + ))); } tracing::info!( "Finished extracting pip tar {folder}. Took {}ms", @@ -139,31 +158,3 @@ pub async fn extract_pip_tar(tar: &str, folder: &str) -> error::Result<()> { ); Ok(()) } - -#[cfg(feature = "enterprise")] -pub async fn execute_command(dir: &str, command: &str, args: Vec<&str>) -> error::Result<()> { - tracing::info!("Executing command: {command} {}", args.iter().join(" ")); - match Command::new(command) - .current_dir(dir) - .args(args.clone()) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .spawn() - { - Ok(mut h) => { - if !h.wait().await.unwrap().success() { - return Err(error::Error::ExecutionErr(format!( - "Failed to apply {command} with args: {}", - args.iter().join(" ") - ))); - } - } - Err(e) => { - return Err(error::Error::ExecutionErr(format!( - "Failed to apply {command} with args: {}. Error: {e:?}", - args.iter().join(" ") - ))); - } - } - Ok(()) -} diff --git a/backend/windmill-worker/src/python_executor.rs b/backend/windmill-worker/src/python_executor.rs index 5e7c7f4962..5d5308430d 100644 --- a/backend/windmill-worker/src/python_executor.rs +++ b/backend/windmill-worker/src/python_executor.rs @@ -47,11 +47,11 @@ const NSJAIL_CONFIG_DOWNLOAD_PY_CONTENT: &str = include_str!("../nsjail/download const NSJAIL_CONFIG_RUN_PYTHON3_CONTENT: &str = include_str!("../nsjail/run.python3.config.proto"); const RELATIVE_PYTHON_LOADER: &str = include_str!("../loader.py"); -#[cfg(feature = "enterprise")] +#[cfg(all(feature = "enterprise", feature = "parquet"))] use crate::global_cache::{build_tar_and_push, pull_from_tar}; -#[cfg(feature = "enterprise")] -use windmill_common::s3_helpers::S3_CACHE_BUCKET; +#[cfg(all(feature = "enterprise", feature = "parquet"))] +use windmill_common::s3_helpers::S3_CACHE_SETTINGS; use crate::{ common::{ @@ -152,7 +152,7 @@ pub async fn pip_compile( write_file(job_dir, file, &requirements).await?; - let mut args = vec!["-q", "--no-header", file, "--resolver=backtracking"]; + let mut args = vec!["-q", "--no-header", file, "--resolver=backtracking", "--strip-extras"]; let mut pip_args = vec![]; let pip_extra_index_url = PIP_EXTRA_INDEX_URL .read() @@ -776,31 +776,89 @@ pub async fn handle_python_reqs( .await?; }; + + let mut req_with_penv: Vec<(String, String)> = vec![]; + for req in requirements { - // todo: handle many reqs let venv_p = format!( "{PIP_CACHE_DIR}/{}", req.replace(' ', "").replace('/', "").replace(':', "") ); if metadata(&venv_p).await.is_ok() { req_paths.push(venv_p); - continue; + } else { + req_with_penv.push((req.to_string(), venv_p)); } + } - #[cfg(feature = "enterprise")] - if let Some(ref bucket) = S3_CACHE_BUCKET.read().await.clone() { + enum PullFromTar { + Pulled(String), + NotPulled(String, String), + } + + #[cfg(all(feature = "enterprise", feature = "parquet"))] + if req_with_penv.len() > 0 { + if let Some(os) = S3_CACHE_SETTINGS.read().await.clone() { if matches!(get_license_plan().await, LicensePlan::Pro) { + append_logs(job_id.clone(), w_id.to_string(), format!("s3 cache not available in Pro Plan"), db).await; tracing::warn!("S3 cache not available in the pro plan"); } else { - sqlx::query_scalar!("UPDATE queue SET last_ping = now() WHERE id = $1", job_id) - .execute(db) - .await?; - if pull_from_tar(bucket, venv_p.clone()).await.is_ok() { - req_paths.push(venv_p.clone()); - continue; + + let (done_tx, mut done_rx) = tokio::sync::mpsc::channel(1); + let job_id_2 = job_id.clone(); + let db_2 = db.clone(); + tokio::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(tokio::time::Duration::from_secs(5)) => { + if let Err(e) = sqlx::query_scalar!("UPDATE queue SET last_ping = now() WHERE id = $1", &job_id_2) + .execute(&db_2) + .await { + tracing::error!("failed to update last_ping: {}", e); + } + } + _ = done_rx.recv() => { + break; + } + } + } + + }); + + let start = std::time::Instant::now(); + let futures = req_with_penv.clone().into_iter().map(|(req, venv_p)| { + let os = os.clone(); + async move { + if pull_from_tar(os, venv_p.clone()).await.is_ok() { + PullFromTar::Pulled(venv_p.to_string()) + } else { + PullFromTar::NotPulled(req.to_string(), venv_p.to_string()) + } + }}).collect::>(); + let results = futures::future::join_all(futures).await; + req_with_penv.clear(); + done_tx.send(()).await.expect("failed to send done"); + let mut pulled = vec![]; + for result in results { + match result { + PullFromTar::Pulled(venv_p) => { + pulled.push(venv_p.split("/").last().unwrap_or_default().to_string()); + req_paths.push(venv_p); + } + PullFromTar::NotPulled(req, venv_p) => { + req_with_penv.push((req, venv_p)); + } + } + } + if pulled.len() > 0 { + append_logs(job_id.clone(), w_id.to_string(), format!("pulled {} from s3 cache in {}ms", pulled.join(", "), start.elapsed().as_millis()), db).await; } } - } + } + } + + for (req, venv_p) in req_with_penv { + let mut logs1 = String::new(); logs1.push_str("\n\n--- PIP INSTALL ---\n"); @@ -931,13 +989,13 @@ pub async fn handle_python_reqs( ); child?; - #[cfg(feature = "enterprise")] - if let Some(bucket) = S3_CACHE_BUCKET.read().await.clone() { + #[cfg(all(feature = "enterprise", feature = "parquet"))] + if let Some(os) = S3_CACHE_SETTINGS.read().await.clone() { if matches!(get_license_plan().await, LicensePlan::Pro) { tracing::warn!("S3 cache not available in the pro plan"); } else { let venv_p = venv_p.clone(); - tokio::spawn(build_tar_and_push(bucket, venv_p)); + tokio::spawn(build_tar_and_push(os, venv_p)); } } req_paths.push(venv_p); diff --git a/frontend/src/lib/components/InstanceSettings.svelte b/frontend/src/lib/components/InstanceSettings.svelte index c39a69f5c8..2e9e0e81b1 100644 --- a/frontend/src/lib/components/InstanceSettings.svelte +++ b/frontend/src/lib/components/InstanceSettings.svelte @@ -23,6 +23,7 @@ import KanidmSetting from '$lib/components/KanidmSetting.svelte' import ZitadelSetting from '$lib/components/ZitadelSetting.svelte' import Password from './Password.svelte' + import S3ConfigSettings from './S3ConfigSettings.svelte' export let tab: string = 'Core' export let hideTabs: boolean = false @@ -494,6 +495,8 @@ placeholder={setting.placeholder} bind:value={values[setting.key]} /> + {:else if setting.fieldType == 's3_config'} + {:else if setting.fieldType == 'number'} + import { Database, Loader2 } from 'lucide-svelte' + import Toggle from './Toggle.svelte' + import { Button } from './common' + import { SettingService } from '$lib/gen' + import { sendUserToast } from '$lib/toast' + import TestConnection from './TestConnection.svelte' + + type BucketConfig = { + bucket: string + region: string + access_key: string + secret_key: string + endpoint: string + } + export let bucket_config: BucketConfig | undefined = undefined + + let loading = false + + async function testConnection() { + loading = true + try { + if (bucket_config) { + await SettingService.testS3Config({ requestBody: bucket_config }) + sendUserToast('Connection successful', false) + } + } catch (e) { + sendUserToast(e.body, true) + } finally { + loading = false + } + } + + +
+ { + if (e.detail) { + bucket_config = { + bucket: '', + region: '', + access_key: '', + secret_key: '', + endpoint: '' + } + } else { + bucket_config = undefined + } + }} + /> +
+{#if bucket_config} +
+ + +
+ + + + + + +{/if} diff --git a/frontend/src/lib/components/TestConnection.svelte b/frontend/src/lib/components/TestConnection.svelte index 78a32d3b58..633e9eeb6f 100644 --- a/frontend/src/lib/components/TestConnection.svelte +++ b/frontend/src/lib/components/TestConnection.svelte @@ -7,8 +7,10 @@ import { workspaceStore } from '$lib/stores' import { tryEvery } from '$lib/utils' + export let workspaceOverride: string | undefined = undefined export let resourceType: string | undefined export let args: Record | any = {} + export let buttonTextOverride: string | undefined = undefined const scripts: { [key: string]: { @@ -81,6 +83,29 @@ export async function main(s3: S3) { return testResult } } + }, + s3_bucket: { + code: ` + +const process = require('process'); + +export async function main(bucket: any) { + const req = await fetch(process.env.BASE_URL + '/api/settings/test_s3_config', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: 'Bearer ' + process.env.WM_TOKEN, + }, + body: JSON.stringify(bucket), + }); + if (!req.ok) { + throw new Error(await req.text()); + } + return await req.text(); +} +`, + lang: 'bun', + argName: 'bucket' } } @@ -92,7 +117,7 @@ export async function main(s3: S3) { const resourceScript = scripts[resourceType] const job = await JobService.runScriptPreview({ - workspace: $workspaceStore!, + workspace: workspaceOverride ?? $workspaceStore!, requestBody: { path: `testConnection: ${resourceType}`, language: resourceScript.lang as Preview.language, @@ -106,7 +131,7 @@ export async function main(s3: S3) { tryEvery({ tryCode: async () => { let testResult = await JobService.getCompletedJob({ - workspace: $workspaceStore!, + workspace: workspaceOverride ?? $workspaceStore!, id: job }) if (resourceScript.additionalCheck) { @@ -128,7 +153,7 @@ export async function main(s3: S3) { ) try { await JobService.cancelQueuedJob({ - workspace: $workspaceStore!, + workspace: workspaceOverride ?? $workspaceStore!, id: job, requestBody: { reason: @@ -159,6 +184,6 @@ export async function main(s3: S3) { {:else} {/if} - Test connection + {buttonTextOverride ?? 'Test connection'} {/if} diff --git a/frontend/src/lib/components/build_workers.ts b/frontend/src/lib/components/build_workers.ts index 43e15f335f..27c4fadd56 100644 --- a/frontend/src/lib/components/build_workers.ts +++ b/frontend/src/lib/components/build_workers.ts @@ -1,6 +1,7 @@ import type { Environment } from 'monaco-editor/esm/vs/editor/editor.api.js' import cssWorker from 'monaco-editor/esm/vs/language/css/css.worker?worker' import htmlWorker from 'monaco-editor/esm/vs/language/html/html.worker?worker' + import jsonWorker from 'monaco-editor/esm/vs/language/json/json.worker?worker' import editorWorker from 'monaco-editor/esm/vs/editor/editor.worker?worker' diff --git a/frontend/src/lib/components/instanceSettings.ts b/frontend/src/lib/components/instanceSettings.ts index 691033e691..112f9a388a 100644 --- a/frontend/src/lib/components/instanceSettings.ts +++ b/frontend/src/lib/components/instanceSettings.ts @@ -16,6 +16,7 @@ export interface Setting { | 'seconds' | 'email' | 'license_key' + | 's3_config' storage: SettingStorage isValid?: (value: any) => boolean error?: string @@ -94,11 +95,10 @@ export const settings: Record = { ee_only: '' }, { - label: 'S3 Bucket for Python Distributed Cache & Large Logs', - description: - 'Bucket to store large logs and cache for distributed python jobs. Format: s3://bucket-name', - key: 's3_cache_bucket', - fieldType: 'text', + label: 'S3 for Python Cache & Large Logs', + description: 'Bucket to store large logs and cache for distributed python jobs.', + key: 's3_cache_config', + fieldType: 's3_config', storage: 'setting', ee_only: '' },