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

feat: Support for Azure Blob storage (#3130)

Browse Source
This commit is contained in:
Guillaume Bouvignies
2024-02-02 19:55:07 +01:00
committed by GitHub
parent 1a1757f9fd
commit bbda7cf268
12 changed files with 447 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.toml
+1 -1
View File
@@ -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" }
backend/windmill-api/openapi-deref.yaml
+18 -1
View File
@@ -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
backend/windmill-api/openapi.yaml
+13 -1
View File
@@ -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
backend/windmill-api/src/job_helpers.rs
+198 -76
View File
@@ -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<String>,
Json(query): Json<DuckdbConnectionSettingsQueryV2>,
) -> JsonResult<DuckdbConnectionSettingsResponse> {
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<String>,
Json(query): Json<PolarsConnectionSettingsQueryV2>,
) -> JsonResult<PolarsConnectionSettingsResponse> {
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<String>,
Json(query): Json<S3ResourceInfoQuery>,
) -> JsonResult<S3Resource> {
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<StorageResourceType>,
pub s3_resource_path: Option<String>,
}
@@ -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<String>, // if none, the file will be placed in windmill_uploads/ with a random name.
pub file_extension: Option<String>, // 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<StorageResourceType>,
pub s3_resource_path: Option<String>, // 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<UserDB>,
token: &str,
w_id: &str,
) -> error::Result<(Option<bool>, Option<S3Resource>)> {
) -> error::Result<(Option<bool>, Option<ObjectStoreResource>)> {
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<UserDB>,
token: &str,
w_id: &str,
s3_resource_path: &str,
) -> error::Result<Option<S3Resource>> {
resource_type: StorageResourceType,
resource_path: &str,
) -> error::Result<Option<ObjectStoreResource>> {
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::<S3Resource>(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::<S3Resource>(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::<AzureBlobResource>(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<String> {
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<u32>,
offset: Option<i64>,
sort: Option<(String, bool)>,
search: Option<(String, String)>,
) -> error::Result<Box<RawValue>> {
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<dyn ObjectStore>,
object_store_client: Arc<dyn ObjectStore>,
file_key: &str,
length: usize,
separator: Option<String>,
has_header: Option<bool>,
) -> error::Result<String> {
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<dyn ObjectStore>,
object_store_client: Arc<dyn ObjectStore>,
file_key: &str,
length: usize,
separator: Option<String>,
@@ -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())
backend/windmill-api/src/resources.rs
+11 -3
View File
@@ -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::<Vec<&str>>();
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 {
backend/windmill-common/src/s3_helpers.rs
+104 -22
View File
@@ -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<bool>,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct AzureBlobStorage {
pub azure_blob_resource_path: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub public_resource: Option<bool>,
}
#[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<u16>,
}
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct AzureBlobResource {
#[serde(rename = "endpoint")]
pub endpoint: Option<String>,
#[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<String>,
}
#[derive(Deserialize, Clone)]
pub struct S3Object {
pub s3: String,
}
pub async fn get_etag_or_empty(s3_resource: &S3Resource, s3_object: S3Object) -> Option<String> {
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<String> {
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<u16>) -> 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<Arc<dyn ObjectStore>> {
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<Arc<dyn ObjectStore>> {
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<Arc<dyn ObjectStore>> {
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));
}
backend/windmill-worker/src/common.rs
+36 -14
View File
@@ -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<S3Resource> {
) -> Option<ObjectStoreResource> {
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::<LargeFileStorage>(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::<S3Resource>(&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::<S3Resource>(
&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::<AzureBlobResource>(
&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<S3Resource> = if s3_etags.is_empty() {
let object_store_resource_opt: Option<ObjectStoreResource> = 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;
frontend/src/lib/components/ResourcePicker.svelte
+16 -5
View File
@@ -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}
frontend/src/lib/components/S3FilePicker.svelte
-1
View File
@@ -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 ||
frontend/src/lib/components/apps/components/inputs/AppS3FileInput.svelte
+2 -1
View File
@@ -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) => {
frontend/src/lib/components/common/fileUpload/FileUpload.svelte
+5 -3
View File
@@ -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)
frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
+43 -23
View File
@@ -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<void> {
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}
<div class="mt-5 flex gap-1">
{#key s3ResourceSettings.s3ResourcePath}
<ResourcePicker resourceType="s3" bind:value={s3ResourceSettings.s3ResourcePath} />
{#key s3ResourceSettings.resourcePath}
<ResourcePicker
resourceType="s3,azure_blob"
bind:value={s3ResourceSettings.resourcePath}
bind:valueType={s3ResourceSettings.resourceType}
/>
{/key}
<Button
size="sm"
variant="contained"
color="dark"
disabled={emptyString(s3ResourceSettings.s3ResourcePath)}
disabled={emptyString(s3ResourceSettings.resourcePath)}
on:click={async () => {
if ($workspaceStore) {
s3FileViewer?.open?.(undefined)
@@ -791,7 +811,7 @@
</div>
<div class="flex flex-col mt-5 mb-1 gap-1">
<Toggle
disabled={emptyString(s3ResourceSettings.s3ResourcePath)}
disabled={emptyString(s3ResourceSettings.resourcePath)}
bind:checked={s3ResourceSettings.publicResource}
options={{
right: 'S3 resource details can be accessed by all users of this workspace',
@@ -811,7 +831,7 @@
<div class="flex mt-5 mb-5 gap-1">
<Button
color="blue"
disabled={emptyString(s3ResourceSettings.s3ResourcePath)}
disabled={emptyString(s3ResourceSettings.resourcePath)}
on:click={() => {
editWindmillLFSSettings()
console.log('Saving S3 settings', s3ResourceSettings)