kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: s3 cache config added to instance settings + parallelized tar pulling

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-20 14:18:21 +01:00
parent 458c476f08
commit 174ead0473
20 changed files with 576 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
Dockerfile
View File

@@ -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; \

4
backend/.cargo/config.toml
View File

@@ -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",

2
backend/Cargo.lock generated
View File

@@ -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",

1
backend/Cargo.toml
View File

@@ -235,3 +235,4 @@ aws-config = "^1"
aws-sdk-sts = "^1"
crc = "^3"
tar = "^0"

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
aa7768505b23651069ae2b18470edc5c0023c45a
73ebc9cb3de7b027f238d60c8ec19da16ae2dc94

9
backend/src/main.rs
View File

@@ -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

162
backend/src/monitor.rs
View File

@@ -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::<S3Settings>(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::<String>(q.value.clone()) {
if let Ok(v) = serde_json::from_value::<String>(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<T: FromStr + DeserializeOwned>(
tracing::error!("Error reloading setting {}: {:?}", setting_name, e)
}
}
async fn load_value_from_global_settings(db: &DB, setting_name: &str) -> error::Result<Option<serde_json::Value>> {
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<T: FromStr + DeserializeOwned>(
db: &DB,
setting_name: &str,
std_env_var: &str,
lock: Arc<RwLock<Option<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()
.and_then(|x| x.parse::<T>().ok());
if let Some(q) = q {
if let Ok(v) = serde_json::from_value::<T>(q.value.clone()) {
if let Ok(v) = serde_json::from_value::<T>(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<T: FromStr + DeserializeOwned + Display>(
lock: Arc<RwLock<T>>,
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<T: FromStr + DeserializeOwned + Display>(
.unwrap_or(default);
if let Some(q) = q {
if let Ok(v) = serde_json::from_value::<T>(q.value.clone()) {
if let Ok(v) = serde_json::from_value::<T>(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::<String>(q.value.clone()) {
if let Ok(v) = serde_json::from_value::<String>(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::<Option<HashMap<String, OAuthClient>>>(q.value.clone())
serde_json::from_value::<Option<HashMap<String, OAuthClient>>>(q.clone())
{
v
} else {
tracing::error!(
"Could not parse oauth setting as a json, found: {:#?}",
&q.value
&q
);
None
}

32
backend/windmill-api/openapi.yaml
View File

@@ -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

34
backend/windmill-api/src/settings.rs
View File

@@ -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<DB>,
authed: ApiAuthed,
Json(test_s3_bucket): Json<S3Settings>,
) -> error::Result<String> {
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,

2
backend/windmill-common/src/global_settings.rs
View File

@@ -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";

102
backend/windmill-common/src/s3_helpers.rs
View File

@@ -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<RwLock<Option<String>>> = Arc::new(RwLock::new(None));
pub static ref S3_CACHE_SETTINGS: Arc<RwLock<Option<Arc<dyn ObjectStore>>>> = 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<bool>,
pub token: Option<String>,
pub port: Option<u16>,
}
#[derive(Debug, Deserialize, Serialize, Clone)]
@@ -163,15 +166,20 @@ pub fn build_object_store_client(
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) => {
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<Arc<dyn ObjectStore>> {
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<DefaultCredentialsChain>) -> error::Result<Arc<dyn ObjectStore>> {
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<Arc<dyn Object
s3_resource.bucket.clone(),
);
let mut store_builder = AmazonS3Builder::new()
.with_client_options(ClientOptions::new().with_timeout_disabled()) // TODO: make it configurable maybe
.with_region(s3_resource.region)
.with_bucket_name(s3_resource.bucket)
.with_endpoint(endpoint);
if let Some(credentials_provider) = credential_providers {
store_builder = store_builder.with_credentials(Arc::new(AwsCredentialAdapter {
inner: credentials_provider,
}));
}
if !s3_resource.use_ssl {
store_builder = store_builder.with_allow_http(true)
}
@@ -281,3 +297,75 @@ fn build_azure_blob_client(
return Ok(Arc::new(store));
}
#[derive(Serialize, Deserialize)]
#[serde(tag = "typ", content = "value")]
pub enum ObjectStoreSettings {
S3(S3Settings),
}
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct S3Settings {
pub bucket: Option<String>,
pub region: Option<String>,
pub access_key: Option<String>,
pub secret_key: Option<String>,
pub endpoint: Option<String>,
pub allow_http: Option<bool>, // default to true
pub store_logs: Option<bool>,
}
fn none_if_empty(s: Option<String>) -> Option<String> {
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<Arc<dyn ObjectStore>> {
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<Arc<Self::Credential>> {
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()),
}))
}
}

6
backend/windmill-worker/Cargo.toml
View File

@@ -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
deno_core.workspace = true

12
backend/windmill-worker/src/bun_executor.rs
View File

@@ -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?;

173
backend/windmill-worker/src/global_cache.rs
View File

@@ -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<dyn ObjectStore>, 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<dyn ObjectStore>, 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(())
}

94
backend/windmill-worker/src/python_executor.rs
View File

@@ -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::<Vec<_>>();
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);

3
frontend/src/lib/components/InstanceSettings.svelte
View File

@@ -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'}
<S3ConfigSettings bind:bucket_config={values[setting.key]} />
{:else if setting.fieldType == 'number'}
<input
type="number"

110
frontend/src/lib/components/S3ConfigSettings.svelte Normal file
View File

@@ -0,0 +1,110 @@
<script lang="ts">
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
}
}
</script>
<div>
<Toggle
options={{ right: 'Enable' }}
checked={Boolean(bucket_config)}
on:change={(e) => {
if (e.detail) {
bucket_config = {
bucket: '',
region: '',
access_key: '',
secret_key: '',
endpoint: ''
}
} else {
bucket_config = undefined
}
}}
/>
</div>
{#if bucket_config}
<div class="flex gap-2">
<Button
spacingSize="sm"
size="xs"
btnClasses="h-8"
color="light"
variant="border"
on:click={testConnection}
>
{#if loading}
<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
{:else}
<Database class="mr-2 !h-4 !w-4" />
{/if}
Test from a server
</Button>
<TestConnection
args={bucket_config}
resourceType="s3_bucket"
workspaceOverride="admins"
buttonTextOverride="Test from a worker"
/>
</div>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Bucket</span>
<input type="text" placeholder="bucket-name" bind:value={bucket_config.bucket} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Region</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_REGION</span
>
<input type="text" bind:value={bucket_config.region} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Access Key ID</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_ACCESS_KEY_ID, pod or ec2 profile</span
>
<input type="text" bind:value={bucket_config.access_key} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Secret Key</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_SECRET_KEY, pod or ec2 profile</span
>
<input type="text" bind:value={bucket_config.secret_key} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Endpoint</span>
<span class="text-tertiary text-2xs">Only needed for non AWS S3 providers like R2 or MinIo</span
>
<input type="text" bind:value={bucket_config.endpoint} />
</label>
{/if}

33
frontend/src/lib/components/TestConnection.svelte
View File

@@ -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<string, any> | 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}
<Database class="mr-2 !h-4 !w-4" />
{/if}
Test connection
{buttonTextOverride ?? 'Test connection'}
</Button>
{/if}

1
frontend/src/lib/components/build_workers.ts
View File

@@ -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'

10
frontend/src/lib/components/instanceSettings.ts
View File

@@ -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<string, Setting[]> = {
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: ''
},