diff --git a/backend/Cargo.toml b/backend/Cargo.toml index c95552fa15..6fc84da90a 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -213,7 +213,7 @@ candle-nn = "0.3.0" tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono", "sql-browser-tokio"] } pin-project = "1" -polars = { version = "0.35.4", features = ["lazy", "parquet", "aws", "csv", "dtype-full", "serde", "strings", "extract_groups"] } +polars = { version = "0.35.4", features = ["lazy", "parquet", "aws", "azure", "csv", "dtype-full", "serde", "strings", "extract_groups"] } polars-io = { version = "0.35.4", features = ["csv"] } object_store = { version = "0.8.0", features = ["aws", "azure"] } openidconnect = { version = "3.4.0" } diff --git a/backend/windmill-api/openapi-deref.yaml b/backend/windmill-api/openapi-deref.yaml index 6373cf032b..f01b3b9c8c 100644 --- a/backend/windmill-api/openapi-deref.yaml +++ b/backend/windmill-api/openapi-deref.yaml @@ -1,6 +1,6 @@ openapi: 3.0.3 info: - version: 1.258.4 + version: 1.259.1 title: Windmill API contact: name: Windmill Team @@ -1524,8 +1524,11 @@ paths: type: string enum: - S3Storage + - AzureBlobStorage s3_resource_path: type: string + azure_blob_resource_path: + type: string public_resource: type: boolean git_sync: @@ -10376,6 +10379,10 @@ paths: in: query schema: type: string + - name: prefix + in: query + schema: + type: string responses: '200': description: List of file keys @@ -10616,6 +10623,11 @@ paths: required: false schema: type: string + - name: resource_type + in: query + required: false + schema: + type: string requestBody: description: File content required: true @@ -10657,6 +10669,11 @@ paths: required: false schema: type: string + - name: resource_type + in: query + required: false + schema: + type: string responses: '200': description: Chunk of the downloaded file diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 87d7e9878f..2a9232f1fe 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -7175,6 +7175,11 @@ paths: required: false schema: type: string + - name: resource_type + in: query + required: false + schema: + type: string requestBody: description: File content required: true @@ -7214,6 +7219,11 @@ paths: required: false schema: type: string + - name: resource_type + in: query + required: false + schema: + type: string responses: "200": description: Chunk of the downloaded file @@ -9389,9 +9399,11 @@ components: properties: type: type: string - enum: ["S3Storage"] + enum: ["S3Storage", "AzureBlobStorage"] s3_resource_path: type: string + azure_blob_resource_path: + type: string public_resource: type: boolean diff --git a/backend/windmill-api/src/job_helpers.rs b/backend/windmill-api/src/job_helpers.rs index 1c12022c06..51b1c3606a 100644 --- a/backend/windmill-api/src/job_helpers.rs +++ b/backend/windmill-api/src/job_helpers.rs @@ -17,6 +17,7 @@ use axum::{ }; use futures::{StreamExt, TryStreamExt}; use hyper::http; +use object_store::azure::AzureConfigKey; use object_store::{ClientConfigKey, ObjectStore}; use polars::{ io::{ @@ -35,6 +36,7 @@ use tokio::io::{copy, AsyncWriteExt}; use tokio_util::io::StreamReader; use tower_http::cors::{Any, CorsLayer}; use windmill_common::error::{Error, JsonResult}; +use windmill_common::s3_helpers::{AzureBlobResource, ObjectStoreResource}; use windmill_common::worker::to_raw_value; use windmill_common::{ db::UserDB, @@ -160,7 +162,7 @@ async fn duckdb_connection_settings_v2( Path(w_id): Path, Json(query): Json, ) -> JsonResult { - let s3_resource_opt = match query.s3_resource_path { + let object_store_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( &authed, @@ -168,6 +170,7 @@ async fn duckdb_connection_settings_v2( Some(user_db), &token, &w_id, + StorageResourceType::S3, // for now we only support S3 for duckdb s3_resource_path.as_str(), ) .await? @@ -178,14 +181,21 @@ async fn duckdb_connection_settings_v2( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(Error::NotFound( + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; - return duckdb_connection_settings( - Path(w_id), - Json(DuckdbConnectionSettingsQuery { s3_resource }), - ) - .await; + match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => { + duckdb_connection_settings( + Path(w_id), + Json(DuckdbConnectionSettingsQuery { s3_resource }), + ) + .await + } + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig( + "DuckDB only works with an S3 storage, Azure Blob is not supported yet".to_string(), + )), + } } #[derive(Deserialize)] @@ -205,7 +215,7 @@ async fn polars_connection_settings( let s3_resource = query.s3_resource; let response = S3fsArgs { - endpoint_url: render_endpoint(&s3_resource), + endpoint_url: render_endpoint(s3_resource.endpoint, s3_resource.use_ssl, s3_resource.port), key: s3_resource.access_key, secret: s3_resource.secret_key, use_ssl: s3_resource.use_ssl, @@ -257,7 +267,7 @@ async fn polars_connection_settings_v2( Path(w_id): Path, Json(query): Json, ) -> JsonResult { - let s3_resource_opt = match query.s3_resource_path { + let object_store_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( &authed, @@ -265,6 +275,7 @@ async fn polars_connection_settings_v2( Some(user_db), &token, &w_id, + StorageResourceType::S3, // for now we only support S3 for polars s3_resource_path.as_str(), ) .await? @@ -275,9 +286,17 @@ async fn polars_connection_settings_v2( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(Error::NotFound( + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; + + let s3_resource = match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => Ok(s3_resource), + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig( + "Polars only works with an S3 storage, Azure Blob is not supported yet".to_string(), + )), + }?; + let s3fs = polars_connection_settings( Path(w_id), Json(PolarsConnectionSettingsQuery { s3_resource: s3_resource.clone() }), @@ -287,7 +306,11 @@ async fn polars_connection_settings_v2( let response = PolarsConnectionSettingsResponse { s3fs_args: s3fs, storage_options: PolarsStorageOptions { - aws_endpoint_url: render_endpoint(&s3_resource), + aws_endpoint_url: render_endpoint( + s3_resource.endpoint, + s3_resource.use_ssl, + s3_resource.port, + ), aws_access_key_id: s3_resource.access_key, aws_secret_access_key: s3_resource.secret_key, aws_region: s3_resource.region, @@ -310,7 +333,7 @@ async fn s3_resource_info( Path(w_id): Path, Json(query): Json, ) -> JsonResult { - let s3_resource_opt = match query.s3_resource_path { + let object_store_resource_opt = match query.s3_resource_path { Some(s3_resource_path) => { get_s3_resource( &authed, @@ -318,6 +341,7 @@ async fn s3_resource_info( Some(user_db), &token, &w_id, + StorageResourceType::S3, s3_resource_path.as_str(), ) .await? @@ -328,10 +352,13 @@ async fn s3_resource_info( s3_resource_opt } }; - let s3_resource = s3_resource_opt.ok_or(Error::NotFound( + let object_store_resource = object_store_resource_opt.ok_or(Error::NotFound( "No datasets storage resource defined at the workspace level".to_string(), ))?; - return Ok(Json(s3_resource)); + match object_store_resource { + ObjectStoreResource::S3Resource(s3_resource) => Ok(Json(s3_resource)), + ObjectStoreResource::AzureBlobResource(_) => Err(Error::BadConfig("Requested S3 resource info but the resource path pointed to an Azure Blob resource type".to_string())) + } } #[derive(Serialize, Deserialize, Clone)] @@ -724,6 +751,8 @@ async fn move_s3_file( #[derive(Deserialize)] struct DownloadFileQuery { pub file_key: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub resource_type: Option, pub s3_resource_path: Option, } @@ -754,6 +783,7 @@ async fn download_s3_file( Some(user_db), &token, &w_id, + query.resource_type.unwrap_or(StorageResourceType::S3), s3_resource_path.as_str(), ) .await? @@ -785,6 +815,8 @@ async fn download_s3_file( struct UploadFileQuery { pub file_key: Option, // if none, the file will be placed in windmill_uploads/ with a random name. pub file_extension: Option, // preferred extension for the file in case a random name has to be generated + #[serde(skip_serializing_if = "Option::is_none")] + pub resource_type: Option, pub s3_resource_path: Option, // custom S3 resource to use for this upload. It None, the workspace S3 resource will be used } @@ -907,6 +939,7 @@ async fn upload_s3_file( Some(user_db), &token, &w_id, + query.resource_type.unwrap_or(StorageResourceType::S3), s3_resource_path.as_str(), ) .await? @@ -975,7 +1008,7 @@ async fn get_workspace_s3_resource<'c>( user_db: Option, token: &str, w_id: &str, -) -> error::Result<(Option, Option)> { +) -> error::Result<(Option, Option)> { let raw_lfs_opt = sqlx::query_scalar!( "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", w_id @@ -1000,37 +1033,46 @@ async fn get_workspace_s3_resource<'c>( "Could not deserialize LargeFileStorage value found in database".to_string(), ) })?; - let s3_lfs = match large_file_storage { - LargeFileStorage::S3Storage(s3_lfs) => s3_lfs, + let (resource_type, public_resource, resource_path) = match large_file_storage { + LargeFileStorage::S3Storage(s3_lfs) => ( + StorageResourceType::S3, + s3_lfs.public_resource, + s3_lfs.s3_resource_path, + ), + LargeFileStorage::AzureBlobStorage(azure_lfs) => ( + StorageResourceType::AzureBlob, + azure_lfs.public_resource, + azure_lfs.azure_blob_resource_path, + ), }; // if the resource is declared public, we replace user_db with None such that the resource info will be // retrieved using `db` (and ACLs will be bypassed) - let effective_user_db = if user_db.is_some() && s3_lfs.public_resource.unwrap_or(false) { + let effective_user_db = if user_db.is_some() && public_resource.unwrap_or(false) { None } else { user_db }; - let stripped_resource_path = match s3_lfs.s3_resource_path.strip_prefix("$res:") { - Some(stripped) => stripped, - None => s3_lfs.s3_resource_path.as_str(), - }; + let stripped_resource_path = resource_path + .strip_prefix("$res:") + .unwrap_or(resource_path.as_str()); let s3_resource = match get_s3_resource( authed, db, effective_user_db, token, w_id, + resource_type, stripped_resource_path, ) .await { Ok(s3_resource) => Ok(s3_resource), - Err(Error::NotAuthorized(_)) if !s3_lfs.public_resource.unwrap_or(false) => Ok(None), + Err(Error::NotAuthorized(_)) if !public_resource.unwrap_or(false) => Ok(None), Err(err) => Err(err), }; - return s3_resource.map(|res| (s3_lfs.public_resource, res)); + return s3_resource.map(|res| (public_resource, res)); } async fn get_s3_resource<'c>( @@ -1039,14 +1081,15 @@ async fn get_s3_resource<'c>( user_db: Option, token: &str, w_id: &str, - s3_resource_path: &str, -) -> error::Result> { + resource_type: StorageResourceType, + resource_path: &str, +) -> error::Result> { let s3_resource_value_raw = get_resource_value_interpolated_internal( authed, user_db, db, w_id, - s3_resource_path, + resource_path, None, token, ) @@ -1056,39 +1099,101 @@ async fn get_s3_resource<'c>( return Err(Error::NotFound("Resource not found".to_string())); } - let s3_resource = serde_json::from_value::(s3_resource_value_raw.unwrap()) - .map_err(|err| { - tracing::error!("Error deserializing S3 resource: {:?}", err); - Error::InternalErr(format!("Error reading s3 resource: {}", err.to_string())) - })?; - return Ok(Some(s3_resource)); + let object_store_resource = match resource_type { + StorageResourceType::S3 => { + let s3_resource = serde_json::from_value::(s3_resource_value_raw.unwrap()) + .map_err(|err| { + tracing::error!("Error deserializing S3 resource: {:?}", err); + Error::InternalErr(format!("Error reading s3 resource: {}", err.to_string())) + })?; + ObjectStoreResource::S3Resource(s3_resource) + } + StorageResourceType::AzureBlob => { + let azure_blob_resource = + serde_json::from_value::(s3_resource_value_raw.unwrap()) + .map_err(|err| { + tracing::error!("Error deserializing S3 resource: {:?}", err); + Error::InternalErr(format!( + "Error reading s3 resource: {}", + err.to_string() + )) + })?; + ObjectStoreResource::AzureBlobResource(azure_blob_resource) + } + }; + return Ok(Some(object_store_resource)); } -fn build_polars_s3_config(s3_resource_ref: &S3Resource) -> CloudOptions { - let s3_resource = s3_resource_ref.to_owned(); - let mut s3_configs: Vec<(AmazonS3ConfigKey, String)> = vec![ - (AmazonS3ConfigKey::Region, s3_resource.region), - (AmazonS3ConfigKey::Bucket, s3_resource.bucket), - ( - AmazonS3ConfigKey::Endpoint, - render_endpoint(s3_resource_ref), - ), - ( - AmazonS3ConfigKey::Client(ClientConfigKey::AllowHttp), - (!s3_resource.use_ssl).to_string(), - ), - ( - AmazonS3ConfigKey::VirtualHostedStyleRequest, - (!s3_resource.path_style).to_string(), - ), - ]; - if let Some(access_key) = s3_resource.access_key { - s3_configs.push((AmazonS3ConfigKey::AccessKeyId, access_key)); +#[derive(Deserialize)] +pub enum StorageResourceType { + S3, + AzureBlob, +} + +fn build_polars_s3_config(object_store_resource_ref: &ObjectStoreResource) -> CloudOptions { + match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource_ref) => { + let s3_resource = s3_resource_ref.to_owned(); + let mut s3_configs: Vec<(AmazonS3ConfigKey, String)> = vec![ + (AmazonS3ConfigKey::Region, s3_resource.region), + (AmazonS3ConfigKey::Bucket, s3_resource.bucket), + ( + AmazonS3ConfigKey::Endpoint, + render_endpoint(s3_resource.endpoint, s3_resource.use_ssl, s3_resource.port), + ), + ( + AmazonS3ConfigKey::Client(ClientConfigKey::AllowHttp), + (!s3_resource.use_ssl).to_string(), + ), + ( + AmazonS3ConfigKey::VirtualHostedStyleRequest, + (!s3_resource.path_style).to_string(), + ), + ]; + if let Some(access_key) = s3_resource.access_key { + if access_key != "" { + s3_configs.push((AmazonS3ConfigKey::AccessKeyId, access_key)); + } + } + if let Some(secret_key) = s3_resource.secret_key { + if secret_key != "" { + s3_configs.push((AmazonS3ConfigKey::SecretAccessKey, secret_key)); + } + } + CloudOptions::default().with_aws(s3_configs) + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + let azure_blob_resource = azure_blob_resource.to_owned(); + let mut azure_blob_configs: Vec<(AzureConfigKey, String)> = vec![ + ( + AzureConfigKey::AccountName, + azure_blob_resource.account_name, + ), + ( + AzureConfigKey::ContainerName, + azure_blob_resource.container_name, + ), + ( + AzureConfigKey::Client(ClientConfigKey::AllowHttp), + (!azure_blob_resource.use_ssl).to_string(), + ), + ]; + if let Some(endpoint) = azure_blob_resource.endpoint { + if endpoint != "" { + azure_blob_configs.push(( + AzureConfigKey::Endpoint, + render_endpoint(endpoint, azure_blob_resource.use_ssl, None), + )) + } + } + if let Some(access_key) = azure_blob_resource.access_key { + if access_key != "" { + azure_blob_configs.push((AzureConfigKey::AccessKey, access_key)); + } + } + CloudOptions::default().with_azure(azure_blob_configs) + } } - if let Some(secret_key) = s3_resource.secret_key { - s3_configs.push((AmazonS3ConfigKey::SecretAccessKey, secret_key)); - } - return CloudOptions::default().with_aws(s3_configs); } async fn read_object_chunk( @@ -1128,10 +1233,10 @@ async fn read_s3_text_object_head( } async fn read_s3_parquet_object_head( - s3_resource_ref: &S3Resource, + object_store_resource_ref: &ObjectStoreResource, file_key: &str, ) -> error::Result { - let s3_cloud_config = build_polars_s3_config(s3_resource_ref); + let polars_cloud_config = build_polars_s3_config(object_store_resource_ref); let args: ScanArgsParquet = ScanArgsParquet { n_rows: Some(1), @@ -1142,14 +1247,19 @@ async fn read_s3_parquet_object_head( low_memory: false, use_statistics: false, hive_partitioning: false, - cloud_options: Some(s3_cloud_config), + cloud_options: Some(polars_cloud_config), }; - let file_key_clone = file_key.to_string(); - let s3_bucket_clone = s3_resource_ref.bucket.to_string(); + let file_key_prefixed = match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource) => { + format!("s3://{}/{}", s3_resource.bucket, file_key).to_string() + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + format!("az://{}/{}", azure_blob_resource.container_name, file_key).to_string() + } + }; let polars_df_result = tokio::task::spawn_blocking(move || { - let s3_file_key = format!("s3://{}/{}", s3_bucket_clone, file_key_clone); - let lzdf_result = LazyFrame::scan_parquet(s3_file_key, args); + let lzdf_result = LazyFrame::scan_parquet(file_key_prefixed, args); match lzdf_result { Err(err) => { tracing::warn!("Error fetching parquet file from S3: {:?}", err); @@ -1174,14 +1284,14 @@ async fn read_s3_parquet_object_head( } async fn read_s3_parquet_chunk( - s3_resource_ref: &S3Resource, + object_store_resource_ref: &ObjectStoreResource, file_key: &str, limit: Option, offset: Option, sort: Option<(String, bool)>, search: Option<(String, String)>, ) -> error::Result> { - let s3_cloud_config = build_polars_s3_config(s3_resource_ref); + let s3_cloud_config = build_polars_s3_config(object_store_resource_ref); let args: ScanArgsParquet = ScanArgsParquet { n_rows: None, @@ -1195,11 +1305,16 @@ async fn read_s3_parquet_chunk( cloud_options: Some(s3_cloud_config), }; - let file_key_clone = file_key.to_string(); - let s3_bucket_clone = s3_resource_ref.bucket.to_string(); + let file_key_prefixed = match object_store_resource_ref { + ObjectStoreResource::S3Resource(s3_resource) => { + format!("s3://{}/{}", s3_resource.bucket, file_key).to_string() + } + ObjectStoreResource::AzureBlobResource(azure_blob_resource) => { + format!("az://{}/{}", azure_blob_resource.container_name, file_key).to_string() + } + }; return tokio::task::spawn_blocking(move || { - let s3_file_key = format!("s3://{}/{}", s3_bucket_clone, file_key_clone); - let lzdf_result = LazyFrame::scan_parquet(s3_file_key, args); + let lzdf_result = LazyFrame::scan_parquet(file_key_prefixed, args); match lzdf_result { Err(err) => { tracing::warn!("Error fetching parquet file from S3: {:?}", err); @@ -1243,26 +1358,33 @@ async fn read_s3_parquet_chunk( })?; } async fn csv_file_preview_with_fallback( - s3_client: Arc, + object_store_client: Arc, file_key: &str, length: usize, separator: Option, has_header: Option, ) -> error::Result { - match read_s3_csv_object_head(s3_client.clone(), &file_key, length, separator, has_header).await + match read_s3_csv_object_head( + object_store_client.clone(), + &file_key, + length, + separator, + has_header, + ) + .await { Ok(csv_preview) => Ok(csv_preview), Err(_) => { // fallback to default text file preview is the CSV could not be parsed. It's a text file after all let raw_text = - read_s3_text_object_head(s3_client.clone(), &file_key, 0, length).await?; + read_s3_text_object_head(object_store_client.clone(), &file_key, 0, length).await?; return Ok(raw_text); } } } async fn read_s3_csv_object_head( - s3_client: Arc, + object_store_client: Arc, file_key: &str, length: usize, separator: Option, @@ -1278,7 +1400,7 @@ async fn read_s3_csv_object_head( }?; let path = object_store::path::Path::from(file_key); - let s3_object = s3_client + let stored_object = object_store_client .get(&path) .await .map_err(|err| { @@ -1290,7 +1412,7 @@ async fn read_s3_csv_object_head( // TODO: polars does not seem to support lazy csv reader, unfortunately. We can implement it ourselves if needed // Right now it's fine b/c we limit the download from AWS to 32MB. We should recommend users to use parquet // for larger files - let file_content_bytes = s3_object + let file_content_bytes = stored_object .take(length as usize) .filter(|obj| future::ready(obj.is_ok())) .map(|obj| obj.unwrap()) diff --git a/backend/windmill-api/src/resources.rs b/backend/windmill-api/src/resources.rs index d6451c4d41..a8eb3dddb8 100644 --- a/backend/windmill-api/src/resources.rs +++ b/backend/windmill-api/src/resources.rs @@ -22,7 +22,7 @@ use bytes::Bytes; use hyper::{header, StatusCode}; use serde::{Deserialize, Serialize}; use serde_json::{value::RawValue, Value}; -use sql_builder::{bind::Bind, SqlBuilder}; +use sql_builder::{bind::Bind, quote, SqlBuilder}; use sqlx::{FromRow, Postgres, Transaction}; use uuid::Uuid; use windmill_audit::{audit_log, ActionKind}; @@ -223,8 +223,16 @@ async fn list_resources( .clone(); if let Some(rt) = &lq.resource_type { - for rt in rt.split(',') { - sqlb.and_where_eq("resource_type", "?".bind(&rt)); + let resource_type_filters = rt.split(',').collect::>(); + if resource_type_filters.len() == 1 { + sqlb.and_where_eq("resource_type", "?".bind(rt)); + } else { + let mut list = Vec::new(); + for rt in resource_type_filters { + let quoted_value = quote(rt); + list.push(quoted_value); + } + sqlb.and_where_in("resource_type", list.as_slice()); } } if let Some(rt) = &lq.resource_type_exclude { diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 2c2b181458..bb666d76fc 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -1,4 +1,5 @@ use crate::error; +use object_store::azure::MicrosoftAzureBuilder; use object_store::ObjectStore; use object_store::{aws::AmazonS3Builder, ClientOptions}; use serde::{Deserialize, Serialize}; @@ -8,6 +9,7 @@ use std::sync::Arc; #[serde(tag = "type")] pub enum LargeFileStorage { S3Storage(S3Storage), + AzureBlobStorage(AzureBlobStorage), // TODO: Add a filesystem type here in the future if needed } @@ -18,6 +20,19 @@ pub struct S3Storage { pub public_resource: Option, } +#[derive(Serialize, Deserialize, Debug)] +pub struct AzureBlobStorage { + pub azure_blob_resource_path: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub public_resource: Option, +} + +#[derive(Clone, Debug)] +pub enum ObjectStoreResource { + S3Resource(S3Resource), + AzureBlobResource(AzureBlobResource), +} + #[derive(Debug, Deserialize, Serialize, Clone)] pub struct S3Resource { #[serde(rename = "bucket")] @@ -36,50 +51,75 @@ pub struct S3Resource { pub port: Option, } +#[derive(Debug, Deserialize, Serialize, Clone)] +pub struct AzureBlobResource { + #[serde(rename = "endpoint")] + pub endpoint: Option, + #[serde(rename = "useSSL")] + pub use_ssl: bool, + #[serde(rename = "accountName")] + pub account_name: String, + #[serde(rename = "containerName")] + pub container_name: String, + #[serde(rename = "accessKey")] + pub access_key: Option, +} + #[derive(Deserialize, Clone)] pub struct S3Object { pub s3: String, } -pub async fn get_etag_or_empty(s3_resource: &S3Resource, s3_object: S3Object) -> Option { - let s3_client = build_object_store_client(s3_resource); - if s3_client.is_err() { +pub async fn get_etag_or_empty( + object_store_resource: &ObjectStoreResource, + s3_object: S3Object, +) -> Option { + let object_store_client = build_object_store_client(object_store_resource); + if object_store_client.is_err() { return None; } - let s3_key = object_store::path::Path::from(s3_object.s3); + let object_key = object_store::path::Path::from(s3_object.s3); - return s3_client + return object_store_client .unwrap() - .head(&s3_key) + .head(&object_key) .await .ok() .map(|meta| meta.e_tag) .flatten(); } -pub fn render_endpoint(s3_resource: &S3Resource) -> String { - let url_with_prefix = if s3_resource.endpoint.starts_with("http://") - || s3_resource.endpoint.starts_with("https://") - { - s3_resource.endpoint.clone() - } else if s3_resource.use_ssl { - format!("https://{}", s3_resource.endpoint) - } else { - format!("http://{}", s3_resource.endpoint) - }; - if s3_resource.port.is_some() { - format!("{}:{}", url_with_prefix, s3_resource.port.unwrap()) +pub fn render_endpoint(raw_endpoint: String, use_ssl: bool, port: Option) -> String { + let url_with_prefix = + if raw_endpoint.starts_with("http://") || raw_endpoint.starts_with("https://") { + raw_endpoint.clone() + } else if use_ssl { + format!("https://{}", raw_endpoint) + } else { + format!("http://{}", raw_endpoint) + }; + if port.is_some() { + format!("{}:{}", url_with_prefix, port.unwrap()) } else { url_with_prefix } } pub fn build_object_store_client( - s3_resource_ref: &S3Resource, + 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) => { + build_azure_blob_client(&azure_blob_resource_ref) + } + } +} + +fn build_s3_client(s3_resource_ref: &S3Resource) -> error::Result> { let s3_resource = s3_resource_ref.clone(); - let endpoint = render_endpoint(&s3_resource); + let endpoint = render_endpoint(s3_resource.endpoint, s3_resource.use_ssl, s3_resource.port); let mut store_builder = AmazonS3Builder::new() .with_client_options(ClientOptions::new().with_timeout_disabled()) // TODO: make it configurable maybe @@ -92,10 +132,14 @@ pub fn build_object_store_client( } if let Some(key) = s3_resource.access_key { - store_builder = store_builder.with_access_key_id(key); + if key != "" { + store_builder = store_builder.with_access_key_id(key); + } } if let Some(secret_key) = s3_resource.secret_key { - store_builder = store_builder.with_secret_access_key(secret_key); + if secret_key != "" { + store_builder = store_builder.with_secret_access_key(secret_key); + } } if !s3_resource.path_style { store_builder = store_builder.with_virtual_hosted_style_request(s3_resource.path_style); @@ -111,3 +155,41 @@ pub fn build_object_store_client( return Ok(Arc::new(store)); } + +fn build_azure_blob_client( + azure_blob_resource_ref: &AzureBlobResource, +) -> error::Result> { + let blob_resource = azure_blob_resource_ref.clone(); + + let mut store_builder = MicrosoftAzureBuilder::new() + .with_client_options(ClientOptions::new().with_timeout_disabled()) // TODO: make it configurable maybe + .with_account(blob_resource.account_name) + .with_container_name(blob_resource.container_name); + + if let Some(endpoint) = blob_resource.endpoint { + if endpoint != "" { + let endpoint = render_endpoint(endpoint, blob_resource.use_ssl, None); + store_builder = store_builder.with_endpoint(endpoint) + } + } + + if !blob_resource.use_ssl { + store_builder = store_builder.with_allow_http(true) + } + + if let Some(key) = blob_resource.access_key { + if key != "" { + store_builder = store_builder.with_access_key(key); + } + } + + let store = store_builder.build().map_err(|err| { + tracing::error!("Error building object store client: {:?}", err); + error::Error::InternalErr(format!( + "Error building object store client: {}", + err.to_string() + )) + })?; + + return Ok(Arc::new(store)); +} diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index e7f907511f..20d3893c1f 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -9,7 +9,10 @@ use serde_json::{json, Value}; use sqlx::{Pool, Postgres}; use tokio::process::Command; use tokio::{fs::File, io::AsyncReadExt}; -use windmill_common::s3_helpers::{get_etag_or_empty, LargeFileStorage, S3Object, S3Resource}; +use windmill_common::s3_helpers::{ + get_etag_or_empty, AzureBlobResource, LargeFileStorage, ObjectStoreResource, S3Object, + S3Resource, +}; use windmill_common::worker::{CLOUD_HOSTED, WORKER_CONFIG}; use windmill_common::{ error::{self, Error}, @@ -935,7 +938,7 @@ async fn get_workspace_s3_resource_path( client: &AuthedClient, workspace_id: &str, job_id: &Uuid, -) -> Option { +) -> Option { let raw_lfs_opt = sqlx::query_scalar!( "SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", workspace_id @@ -947,14 +950,32 @@ async fn get_workspace_s3_resource_path( .map(|val| serde_json::from_value::(val).ok()) .flatten(); - if let Some(LargeFileStorage::S3Storage(s3_storage)) = raw_lfs_opt { - let resource_path = s3_storage.s3_resource_path.trim_start_matches("$res:"); - client - .get_resource_value_interpolated::(&resource_path, Some(job_id.to_string())) - .await - .ok() - } else { - return None; + match raw_lfs_opt { + Some(LargeFileStorage::S3Storage(s3_storage)) => { + let resource_path = s3_storage.s3_resource_path.trim_start_matches("$res:"); + let s3_resource = client + .get_resource_value_interpolated::( + &resource_path, + Some(job_id.to_string()), + ) + .await + .ok(); + s3_resource.map(|resource| ObjectStoreResource::S3Resource(resource)) + } + Some(LargeFileStorage::AzureBlobStorage(azure_blob_storage)) => { + let resource_path = azure_blob_storage + .azure_blob_resource_path + .trim_start_matches("$res:"); + let azure_blob_resource = client + .get_resource_value_interpolated::( + &resource_path, + Some(job_id.to_string()), + ) + .await + .ok(); + azure_blob_resource.map(|resource| ObjectStoreResource::AzureBlobResource(resource)) + } + None => None, } } @@ -1031,19 +1052,20 @@ pub async fn get_cached_resource_value_if_valid( return None; } let s3_etags = cached_resource.s3_etags.unwrap_or_default(); - let s3_resource_opt: Option = if s3_etags.is_empty() { + let object_store_resource_opt: Option = if s3_etags.is_empty() { None } else { get_workspace_s3_resource_path(db, &client, workspace_id, job_id).await }; - if !s3_etags.is_empty() && s3_resource_opt.is_none() { + if !s3_etags.is_empty() && object_store_resource_opt.is_none() { tracing::warn!("Cached result references s3 files that are not retrievable anymore because the workspace S3 resource can't be fetched. Cache will be invalidated"); return None; } for (s3_file_key, s3_file_etag) in s3_etags { - if let Some(s3_resource) = s3_resource_opt.clone() { + if let Some(object_store_resource) = object_store_resource_opt.clone() { let etag = - get_etag_or_empty(&s3_resource, S3Object { s3: s3_file_key.clone() }).await; + get_etag_or_empty(&object_store_resource, S3Object { s3: s3_file_key.clone() }) + .await; if etag.is_none() || etag.clone().unwrap() != s3_file_etag { tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag); return None; diff --git a/frontend/src/lib/components/ResourcePicker.svelte b/frontend/src/lib/components/ResourcePicker.svelte index e2ade97be0..45b459b28c 100644 --- a/frontend/src/lib/components/ResourcePicker.svelte +++ b/frontend/src/lib/components/ResourcePicker.svelte @@ -15,6 +15,7 @@ export let initialValue: string | undefined = undefined export let value: string | undefined = initialValue + export let valueType: string | undefined = undefined export let resourceType: string | undefined = undefined export let disablePortal = false export let showSchemaExplorer = false @@ -23,7 +24,8 @@ initialValue || value ? { value: value ?? initialValue, - label: value ?? initialValue + label: value ?? initialValue, + type: valueType } : undefined @@ -39,12 +41,13 @@ .filter((x) => x.resource_type != 'state' && x.resource_type != 'cache') .map((x) => ({ value: x.path, - label: x.path + label: x.path, + type: x.resource_type })) // TODO check if this is needed if (!nc.find((x) => x.value == value) && (initialValue || value)) { - nc.push({ value: value ?? initialValue!, label: value ?? initialValue! }) + nc.push({ value: value ?? initialValue!, label: value ?? initialValue!, type: '' }) } collection = nc } @@ -68,7 +71,12 @@ on:refresh={async (e) => { await loadResources(resourceType) value = e.detail - valueSelect = { value: e.detail, label: e.detail } + valueType = collection.find((x) => x?.value == value)?.type + valueSelect = { + value: e.detail, + label: e.detail, + type: valueType ?? '' + } }} newPageOAuth bind:this={appConnect} @@ -80,7 +88,8 @@ await loadResources(resourceType) if (e.detail) { value = e.detail - valueSelect = { value: e.detail, label: e.detail } + valueType = collection.find((x) => x?.value == value)?.type + valueSelect = { value: e.detail, label: e.detail, type: valueType ?? '' } } }} /> @@ -91,10 +100,12 @@ value={valueSelect} on:change={(e) => { value = e.detail.value + valueType = e.detail.type valueSelect = e.detail }} on:clear={() => { value = undefined + valueType = undefined valueSelect = undefined }} items={collection} diff --git a/frontend/src/lib/components/S3FilePicker.svelte b/frontend/src/lib/components/S3FilePicker.svelte index 77fba7751a..8197abff5a 100644 --- a/frontend/src/lib/components/S3FilePicker.svelte +++ b/frontend/src/lib/components/S3FilePicker.svelte @@ -118,7 +118,6 @@ marker: page == 0 ? undefined : listMarkers[page - 1], prefix: filter.trim() != '' ? filter : undefined }) - console.log(availableFiles?.windmill_large_files?.length) if ( availableFiles.restricted_access === null || availableFiles.restricted_access === undefined || diff --git a/frontend/src/lib/components/apps/components/inputs/AppS3FileInput.svelte b/frontend/src/lib/components/apps/components/inputs/AppS3FileInput.svelte index 1a06cebd39..28090f133b 100644 --- a/frontend/src/lib/components/apps/components/inputs/AppS3FileInput.svelte +++ b/frontend/src/lib/components/apps/components/inputs/AppS3FileInput.svelte @@ -118,7 +118,8 @@ pathTransformer={resolvedConfig?.type?.configuration?.s3?.pathTemplate} allowMultiple={resolvedConfigS3.allowMultiple} containerText={resolvedConfigS3.text} - customS3ResourcePath={resolvedConfigS3.resource} + customResourcePath={resolvedConfigS3.resource} + customResourceType="s3" customClass={css?.container?.class} customStyle={css?.container?.style} on:addition={(evt) => { diff --git a/frontend/src/lib/components/common/fileUpload/FileUpload.svelte b/frontend/src/lib/components/common/fileUpload/FileUpload.svelte index 4b3f7aeb95..9d18a96275 100644 --- a/frontend/src/lib/components/common/fileUpload/FileUpload.svelte +++ b/frontend/src/lib/components/common/fileUpload/FileUpload.svelte @@ -15,7 +15,8 @@ export let acceptedFileTypes: string[] | undefined = ['*'] export let allowMultiple: boolean = true export let containerText: string = 'Drag and drop files here or click to browse' - export let customS3ResourcePath: string | undefined = undefined + export let customResourcePath: string | undefined = undefined + export let customResourceType: 's3' | 'azure_blob' | undefined = undefined // when customResourcePath is provided, this should be provided as well. Will default to S3 if not export let customClass: string = '' export let customStyle: string = '' export let randomFileKey: boolean = false @@ -99,8 +100,9 @@ if (path) { params.append('file_key', path) } - if (customS3ResourcePath?.split(':')[1]) { - params.append('s3_resource_path', customS3ResourcePath?.split(':')[1]) + if (customResourcePath?.split(':')[1]) { + params.append('s3_resource_path', customResourcePath?.split(':')[1]) + params.append('resource_type', customResourceType === 'azure_blob' ? 'AzureBlob' : 'S3') } if (fileExtension) { params.append('file_extension', fileExtension) diff --git a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte index aa73ac28d9..1fc65098d3 100644 --- a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte @@ -59,7 +59,8 @@ let errorHandlerMutedOnCancel: boolean | undefined = undefined let openaiResourceInitialPath: string | undefined = undefined let s3ResourceSettings: { - s3ResourcePath: string | undefined + resourceType: 's3' | 'azure_blob' + resourcePath: string | undefined publicResource: boolean | undefined } let gitSyncSettings: { @@ -175,16 +176,22 @@ } async function editWindmillLFSSettings(): Promise { - if (!emptyString(s3ResourceSettings.s3ResourcePath)) { - let resourcePathWithPrefix = `$res:${s3ResourceSettings.s3ResourcePath}` + if (!emptyString(s3ResourceSettings.resourcePath)) { + let resourcePathWithPrefix = `$res:${s3ResourceSettings.resourcePath}` + let params = { + public_resource: s3ResourceSettings.publicResource + } + if (s3ResourceSettings.resourceType === 'azure_blob') { + params['type'] = LargeFileStorage.type.AZURE_BLOB_STORAGE + params['azure_blob_resource_path'] = resourcePathWithPrefix + } else { + params['type'] = LargeFileStorage.type.S3STORAGE + params['s3_resource_path'] = resourcePathWithPrefix + } await WorkspaceService.editLargeFileStorageConfig({ workspace: $workspaceStore!, requestBody: { - large_file_storage: { - type: LargeFileStorage.type.S3STORAGE, - s3_resource_path: resourcePathWithPrefix, - public_resource: s3ResourceSettings.publicResource - } + large_file_storage: params } }) sendUserToast(`Large file storage settings updated`) @@ -284,16 +291,25 @@ codeCompletionEnabled = settings.code_completion_enabled workspaceDefaultAppPath = settings.default_app - s3ResourceSettings = - settings.large_file_storage?.type === LargeFileStorage.type.S3STORAGE - ? { - s3ResourcePath: settings.large_file_storage?.s3_resource_path?.replace('$res:', ''), - publicResource: settings.large_file_storage?.public_resource - } - : { - s3ResourcePath: undefined, - publicResource: undefined - } + if (settings.large_file_storage?.type === LargeFileStorage.type.S3STORAGE) { + s3ResourceSettings = { + resourceType: 's3', + resourcePath: settings.large_file_storage?.s3_resource_path?.replace('$res:', ''), + publicResource: settings.large_file_storage?.public_resource + } + } else if (settings.large_file_storage?.type === LargeFileStorage.type.AZURE_BLOB_STORAGE) { + s3ResourceSettings = { + resourceType: 'azure_blob', + resourcePath: settings.large_file_storage?.azure_blob_resource_path?.replace('$res:', ''), + publicResource: settings.large_file_storage?.public_resource + } + } else { + s3ResourceSettings = { + resourceType: 's3', + resourcePath: undefined, + publicResource: undefined + } + } if ( settings.git_sync !== undefined && settings.git_sync !== null && @@ -774,14 +790,18 @@ {/if} {#if s3ResourceSettings}
- {#key s3ResourceSettings.s3ResourcePath} - + {#key s3ResourceSettings.resourcePath} + {/key}