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

generalize cache object store

Browse Source
This commit is contained in:
Ruben Fiszel
2024-03-21 08:50:10 +01:00
parent c1b1da2733
commit 030f6c553e
9 changed files with 49 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+7 -2
View File
@@ -37,7 +37,7 @@ use windmill_common::{
use windmill_common::METRICS_ADDR;
#[cfg(feature = "parquet")]
use windmill_common::global_settings::S3_CACHE_CONFIG_SETTING;
use windmill_common::global_settings::OBJECT_STORE_CACHE_CONFIG_SETTING;
use windmill_worker::{
BUN_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
@@ -448,7 +448,7 @@ Windmill Community Edition {GIT_VERSION}
reload_job_default_timeout_setting(&db).await
},
#[cfg(feature = "parquet")]
S3_CACHE_CONFIG_SETTING => {
OBJECT_STORE_CACHE_CONFIG_SETTING => {
reload_s3_cache_setting(&db).await
},
SCIM_TOKEN_SETTING => {
@@ -562,6 +562,11 @@ Windmill Community Edition {GIT_VERSION}
tracing::info!("Nothing to do, exiting.");
}
tracing::info!("Exiting connection pool");
tokio::select! {
_ = db.close() => {
tracing::info!("Database connection pool closed");
},
}
db.close().await;
Ok(())
}
backend/src/monitor.rs
+9 -7
View File
@@ -40,10 +40,10 @@ use windmill_worker::{
};
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::{build_s3_client_from_settings, S3_CACHE_SETTINGS, S3Settings};
use windmill_common::s3_helpers::{build_object_store_from_settings, build_s3_client_from_settings, OBJECT_STORE_CACHE_SETTINGS, S3Settings};
#[cfg(feature = "parquet")]
use windmill_common::global_settings::S3_CACHE_CONFIG_SETTING;
use windmill_common::global_settings::OBJECT_STORE_CACHE_CONFIG_SETTING;
#[cfg(feature = "enterprise")]
use crate::ee::verify_license_key;
@@ -397,17 +397,19 @@ pub async fn reload_retention_period_setting(db: &DB) {
#[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;
use windmill_common::s3_helpers::ObjectSettings;
let s3_config = load_value_from_global_settings(db, OBJECT_STORE_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);
let mut s3_cache_settings = OBJECT_STORE_CACHE_SETTINGS.write().await;
let setting = serde_json::from_value::<ObjectSettings>(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;
let s3_client = build_object_store_from_settings(setting.unwrap()).await;
if let Err(e) = s3_client {
tracing::error!("Error building s3 client from settings: {:?}", e)
} else {
@@ -415,7 +417,7 @@ pub async fn reload_s3_cache_setting(db: &DB) {
}
}
} else {
let mut s3_cache_settings = S3_CACHE_SETTINGS.write().await;
let mut s3_cache_settings = OBJECT_STORE_CACHE_SETTINGS.write().await;
if std::env::var("S3_CACHE_BUCKET").is_ok() {
*s3_cache_settings = build_s3_client_from_settings(S3Settings {
bucket: None,
backend/windmill-api/src/settings.rs
+4 -6
View File
@@ -101,10 +101,10 @@ pub async fn test_email(
}
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::S3Settings;
use windmill_common::s3_helpers::ObjectSettings;
#[cfg(feature = "parquet")]
use windmill_common::s3_helpers::build_s3_client_from_settings;
use windmill_common::s3_helpers::build_object_store_from_settings;
@@ -112,14 +112,12 @@ use windmill_common::s3_helpers::build_s3_client_from_settings;
pub async fn test_s3_bucket(
Extension(db): Extension<DB>,
authed: ApiAuthed,
Json(test_s3_bucket): Json<S3Settings>,
Json(test_s3_bucket): Json<ObjectSettings>,
) -> 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 client = build_object_store_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}");
backend/windmill-common/src/global_settings.rs
+1 -1
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_CONFIG_SETTING: &str = "s3_cache_config";
pub const OBJECT_STORE_CACHE_CONFIG_SETTING: &str = "object_store_cache_config";
pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation";
backend/windmill-common/src/s3_helpers.rs
+19 -2
View File
@@ -21,7 +21,7 @@ use tokio::sync::RwLock;
#[cfg(feature = "parquet")]
lazy_static::lazy_static! {
pub static ref S3_CACHE_SETTINGS: Arc<RwLock<Option<Arc<dyn ObjectStore>>>> = Arc::new(RwLock::new(None));
pub static ref OBJECT_STORE_CACHE_SETTINGS: Arc<RwLock<Option<Arc<dyn ObjectStore>>>> = Arc::new(RwLock::new(None));
}
#[derive(Serialize, Deserialize, Debug)]
@@ -84,7 +84,6 @@ pub struct S3Resource {
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct AzureBlobResource {
#[serde(rename = "endpoint")]
pub endpoint: Option<String>,
#[serde(rename = "useSSL")]
pub use_ssl: Option<bool>,
@@ -311,6 +310,24 @@ pub enum ObjectStoreSettings {
S3(S3Settings),
}
#[derive(Debug, Deserialize, Serialize, Clone)]
#[serde(tag = "type")]
pub enum ObjectSettings {
S3(S3Settings),
Azure(AzureBlobResource),
}
#[cfg(feature = "parquet")]
pub async fn build_object_store_from_settings(settings: ObjectSettings) -> error::Result<Arc<dyn ObjectStore>> {
match settings {
ObjectSettings::S3(s3_settings) => build_s3_client_from_settings(s3_settings).await,
ObjectSettings::Azure(azure_settings) => {
let azure_blob_resource = azure_settings;
build_azure_blob_client(&azure_blob_resource)
}
}
}
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct S3Settings {
pub bucket: Option<String>,
backend/windmill-worker/src/python_executor.rs
+3 -3
View File
@@ -51,7 +51,7 @@ const RELATIVE_PYTHON_LOADER: &str = include_str!("../loader.py");
use crate::global_cache::{build_tar_and_push, pull_from_tar};
#[cfg(all(feature = "enterprise", feature = "parquet"))]
use windmill_common::s3_helpers::S3_CACHE_SETTINGS;
use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
use crate::{
common::{
@@ -799,7 +799,7 @@ pub async fn handle_python_reqs(
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if req_with_penv.len() > 0 {
if let Some(os) = S3_CACHE_SETTINGS.read().await.clone() {
if let Some(os) = OBJECT_STORE_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");
@@ -991,7 +991,7 @@ pub async fn handle_python_reqs(
child?;
#[cfg(all(feature = "enterprise", feature = "parquet"))]
if let Some(os) = S3_CACHE_SETTINGS.read().await.clone() {
if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
if matches!(get_license_plan().await, LicensePlan::Pro) {
tracing::warn!("S3 cache not available in the pro plan");
} else {
frontend/src/lib/components/InstanceSettings.svelte
+3 -3
View File
@@ -23,7 +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'
import ObjectStoreConfigSettings from './ObjectStoreConfigSettings.svelte'
export let tab: string = 'Core'
export let hideTabs: boolean = false
@@ -495,8 +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 == 'object_store_config'}
<ObjectStoreConfigSettings bind:bucket_config={values[setting.key]} />
{:else if setting.fieldType == 'number'}
<input
type="number"
frontend/src/lib/components/S3ConfigSettings.svelte
-110
View File
@@ -1,110 +0,0 @@
<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}
frontend/src/lib/components/instanceSettings.ts
+3 -3
View File
@@ -16,7 +16,7 @@ export interface Setting {
| 'seconds'
| 'email'
| 'license_key'
| 's3_config'
| 'object_store_config'
storage: SettingStorage
isValid?: (value: any) => boolean
error?: string
@@ -97,8 +97,8 @@ export const settings: Record<string, Setting[]> = {
{
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',
key: 'object_store_cache_config',
fieldType: 'object_store_config',
storage: 'setting',
ee_only: ''
},