From 7c5ea569a8102ef052d42216e2ff8d4c3169a7a5 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Wed, 30 Aug 2023 13:35:41 +0200 Subject: [PATCH] feat: add global cache configuration --- ...e57285ce64ac41a84681b32deb71e452334c1.json | 22 +++++ ...ca217eb19a15a250c3b83ca63f606f89b9681.json | 15 ++++ ...0230829202651_add_global_settings.down.sql | 1 + .../20230829202651_add_global_settings.up.sql | 6 ++ backend/src/main.rs | 69 +-------------- backend/windmill-api/openapi.yaml | 59 +++++++++++++ backend/windmill-api/src/lib.rs | 4 +- backend/windmill-api/src/settings.rs | 83 +++++++++++++++++++ .../windmill-common/src/global_settings.rs | 58 +++++++++++++ backend/windmill-common/src/lib.rs | 1 + backend/windmill-worker/src/global_cache.rs | 24 ++++++ backend/windmill-worker/src/worker.rs | 24 +++--- .../(root)/(logged)/workers/+page.svelte | 40 ++++++++- 13 files changed, 328 insertions(+), 78 deletions(-) create mode 100644 backend/.sqlx/query-1c1577b9963d907c4245a027fece57285ce64ac41a84681b32deb71e452334c1.json create mode 100644 backend/.sqlx/query-a59b70164dc87224d09a04d5469ca217eb19a15a250c3b83ca63f606f89b9681.json create mode 100644 backend/migrations/20230829202651_add_global_settings.down.sql create mode 100644 backend/migrations/20230829202651_add_global_settings.up.sql create mode 100644 backend/windmill-api/src/settings.rs create mode 100644 backend/windmill-common/src/global_settings.rs diff --git a/backend/.sqlx/query-1c1577b9963d907c4245a027fece57285ce64ac41a84681b32deb71e452334c1.json b/backend/.sqlx/query-1c1577b9963d907c4245a027fece57285ce64ac41a84681b32deb71e452334c1.json new file mode 100644 index 0000000000..e1719ead57 --- /dev/null +++ b/backend/.sqlx/query-1c1577b9963d907c4245a027fece57285ce64ac41a84681b32deb71e452334c1.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value FROM global_settings WHERE name = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "1c1577b9963d907c4245a027fece57285ce64ac41a84681b32deb71e452334c1" +} diff --git a/backend/.sqlx/query-a59b70164dc87224d09a04d5469ca217eb19a15a250c3b83ca63f606f89b9681.json b/backend/.sqlx/query-a59b70164dc87224d09a04d5469ca217eb19a15a250c3b83ca63f606f89b9681.json new file mode 100644 index 0000000000..da8842f9a7 --- /dev/null +++ b/backend/.sqlx/query-a59b70164dc87224d09a04d5469ca217eb19a15a250c3b83ca63f606f89b9681.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO global_settings (name, value) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET value = $2, updated_at = now()", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Jsonb" + ] + }, + "nullable": [] + }, + "hash": "a59b70164dc87224d09a04d5469ca217eb19a15a250c3b83ca63f606f89b9681" +} diff --git a/backend/migrations/20230829202651_add_global_settings.down.sql b/backend/migrations/20230829202651_add_global_settings.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230829202651_add_global_settings.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230829202651_add_global_settings.up.sql b/backend/migrations/20230829202651_add_global_settings.up.sql new file mode 100644 index 0000000000..29597cf4bd --- /dev/null +++ b/backend/migrations/20230829202651_add_global_settings.up.sql @@ -0,0 +1,6 @@ +-- Add up migration script here +CREATE TABLE global_settings ( + name VARCHAR(255) PRIMARY KEY, + value JSONB NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); \ No newline at end of file diff --git a/backend/src/main.rs b/backend/src/main.rs index e99ecf2dae..34594c7536 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -20,7 +20,7 @@ use tokio::{ sync::RwLock, }; use windmill_api::{LICENSE_KEY, OAUTH_CLIENTS, SMTP_CLIENT}; -use windmill_common::{utils::rd_string, METRICS_ADDR}; +use windmill_common::{global_settings::ENV_SETTINGS, utils::rd_string, METRICS_ADDR}; use windmill_worker::{ BUN_CACHE_DIR, BUN_TMP_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM, DENO_TMP_CACHE_DIR, DENO_TMP_CACHE_DIR_DEPS, DENO_TMP_CACHE_DIR_NPM, GO_BIN_CACHE_DIR, @@ -48,7 +48,7 @@ async fn main() -> anyhow::Result<()> { .unwrap_or(DEFAULT_NUM_WORKERS as i32); if num_workers > 1 { - tracing::warn!("We recommend using at most 1 worker per container, use more only if you know what you are doing."); + tracing::warn!("We STRONGLY recommend using at most 1 worker per container, unless this worker is dedicated to native jobs only. "); } let metrics_addr: Option = *METRICS_ADDR; @@ -119,68 +119,7 @@ Windmill Community Edition {GIT_VERSION} ##############################" ); - display_config(vec![ - "DISABLE_NSJAIL", - "DISABLE_SERVER", - "NUM_WORKERS", - "METRICS_ADDR", - "JSON_FMT", - "BASE_URL", - "TIMEOUT", - "ZOMBIE_JOB_TIMEOUT", - "RESTART_ZOMBIE_JOBS", - "SLEEP_QUEUE", - "MAX_LOG_SIZE", - "SERVER_BIND_ADDR", - "PORT", - "KEEP_JOB_DIR", - "S3_CACHE_BUCKET", - "TAR_CACHE_RATE", - "COOKIE_DOMAIN", - "PYTHON_PATH", - "DENO_PATH", - "GO_PATH", - "GOPRIVATE", - "GOPROXY", - "NETRC", - "PIP_INDEX_URL", - "PIP_EXTRA_INDEX_URL", - "PIP_TRUSTED_HOST", - "PATH", - "HOME", - "DATABASE_CONNECTIONS", - "TIMEOUT_WAIT_RESULT", - "QUEUE_LIMIT_WAIT_RESULT", - "DENO_AUTH_TOKENS", - "DENO_FLAGS", - "NPM_CONFIG_REGISTRY", - "PIP_LOCAL_DEPENDENCIES", - "ADDITIONAL_PYTHON_PATHS", - "INCLUDE_HEADERS", - "INSTANCE_EVENTS_WEBHOOK", - "CLOUD_HOSTED", - "GLOBAL_CACHE_INTERVAL", - "WORKER_TAGS", - "CUSTOM_TAGS", - "JOB_RETENTION_SECS", - "WAIT_RESULT_FAST_POLL_DURATION_SECS", - "WAIT_RESULT_SLOW_POLL_INTERVAL_MS", - "WAIT_RESULT_FAST_POLL_INTERVAL_MS", - "EXIT_AFTER_NO_JOB_FOR_SECS", - "REQUEST_SIZE_LIMIT", - "SMTP_HOST", - "SMTP_USERNAME", - "SMTP_PORT", - "SMTP_TLS_IMPLICIT", - "CREATE_WORKSPACE_REQUIRE_SUPERADMIN", - "GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE", - ]); - - if std::env::var("WHITELIST_WORKSPACES").is_ok() - || std::env::var("BLACKLIST_WORKSPACES").is_ok() - { - panic!("WHITELIST_WORKSPACES and BLACKLIST_WORKSPACES have been removed, please use Worker Groups instead"); - } + display_config(&ENV_SETTINGS); tracing::info!("Loading OAuth providers...: {:#?}", *OAUTH_CLIENTS); if let Some(ref smtp) = *SMTP_CLIENT { @@ -264,7 +203,7 @@ Windmill Community Edition {GIT_VERSION} Ok(()) } -fn display_config(envs: Vec<&str>) { +fn display_config(envs: &[&str]) { tracing::info!( "config: {}", envs.iter() diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index cf5a5f9c71..5f59e30f12 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -522,6 +522,59 @@ paths: schema: type: boolean + /settings/global/{key}: + get: + summary: get global settings + operationId: getGlobal + tags: + - setting + parameters: + - $ref: "#/components/parameters/Key" + responses: + "200": + description: status + content: + application/json: + schema: {} + + post: + summary: post global settings + operationId: setGlobal + tags: + - setting + parameters: + - $ref: "#/components/parameters/Key" + requestBody: + description: value set + required: true + content: + application/json: + schema: + type: object + properties: + value: {} + + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + + /settings/local: + get: + summary: get local settings + operationId: getLocal + tags: + - setting + responses: + "200": + description: status + content: + application/json: + schema: {} + /users/email: get: summary: get current user email (if logged in) @@ -5450,6 +5503,12 @@ components: name: token parameters: + Key: + name: key + in: path + required: true + schema: + type: string WorkspaceId: name: workspace in: path diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 59d9bb9d01..2c6b4e60bf 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -59,6 +59,7 @@ mod saml; mod schedule; mod scim; mod scripts; +mod settings; mod static_assets; mod tracing_init; mod users; @@ -97,7 +98,7 @@ lazy_static::lazy_static! { .build().unwrap(); pub static ref OAUTH_CLIENTS: AllClients = build_oauth_clients(&BASE_URL) - .map_err(|e| tracing::error!("Error building oauth clients: {}", e)) + .map_err(|e| tracing::error!("Error building oauth clients (is the oauth.json mounted and in correct format? Use '{}' as minimal oauth.json): {}", "{}", e)) .unwrap(); pub static ref SMTP_CLIENT: Option> = { @@ -233,6 +234,7 @@ pub async fn run_server( "/users", users::global_service().layer(Extension(argon2.clone())), ) + .nest("/settings", settings::global_service()) .nest("/jobs", jobs::global_root_service()) .nest("/workers", workers::global_service()) .nest("/scripts", scripts::global_service()) diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs new file mode 100644 index 0000000000..c5e7546fbf --- /dev/null +++ b/backend/windmill-api/src/settings.rs @@ -0,0 +1,83 @@ +/* + * Author: Ruben Fiszel + * Copyright: Windmill Labs, Inc 2023 + * This file and its contents are licensed under the AGPLv3 License. + * Please see the included NOTICE for copyright information and + * LICENSE-AGPL for a copy of the license. + */ + +use crate::{ + db::{ApiAuthed, DB}, + utils::require_super_admin, +}; + +use axum::{ + extract::{Extension, Path}, + routing::{get, post}, + Json, Router, +}; + +use windmill_common::{ + error::{self, JsonResult}, + global_settings::ENV_SETTINGS, +}; + +pub fn global_service() -> Router { + Router::new() + .route("/local", get(get_local_settings)) + .route( + "/global/:key", + post(set_global_setting).get(get_global_setting), + ) +} + +pub async fn get_local_settings( + Extension(db): Extension, + authed: ApiAuthed, +) -> error::JsonResult { + require_super_admin(&db, &authed.email).await?; + + let mut settings = serde_json::Map::new(); + for key in ENV_SETTINGS.iter() { + if let Some(value) = std::env::var(key).ok() { + settings.insert(key.to_string(), serde_json::Value::String(value)); + } + } + Ok(Json(serde_json::Value::Object(settings))) +} + +#[derive(serde::Deserialize)] +pub struct Value { + pub value: serde_json::Value, +} +pub async fn set_global_setting( + Extension(db): Extension, + authed: ApiAuthed, + Path(key): Path, + Json(value): Json, +) -> error::Result<()> { + require_super_admin(&db, &authed.email).await?; + sqlx::query!( + "INSERT INTO global_settings (name, value) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET value = $2, updated_at = now()", + key, + value.value + ) + .execute(&db) + .await?; + tracing::info!("Set global setting {} to {}", key, value.value); + Ok(()) +} + +pub async fn get_global_setting( + Extension(db): Extension, + authed: ApiAuthed, + Path(key): Path, +) -> JsonResult { + require_super_admin(&db, &authed.email).await?; + let value = sqlx::query!("SELECT value FROM global_settings WHERE name = $1", key) + .fetch_optional(&db) + .await? + .map(|x| x.value); + + Ok(Json(value.unwrap_or_else(|| serde_json::Value::Null))) +} diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs new file mode 100644 index 0000000000..44a6567422 --- /dev/null +++ b/backend/windmill-common/src/global_settings.rs @@ -0,0 +1,58 @@ +pub const WORKER_S3_BUCKET_SYNC: &str = "worker_s3_bucket_sync"; + +pub const ENV_SETTINGS: [&str; 54] = [ + "DISABLE_NSJAIL", + "DISABLE_SERVER", + "NUM_WORKERS", + "METRICS_ADDR", + "JSON_FMT", + "BASE_URL", + "TIMEOUT", + "ZOMBIE_JOB_TIMEOUT", + "RESTART_ZOMBIE_JOBS", + "SLEEP_QUEUE", + "MAX_LOG_SIZE", + "SERVER_BIND_ADDR", + "PORT", + "KEEP_JOB_DIR", + "S3_CACHE_BUCKET", + "TAR_CACHE_RATE", + "COOKIE_DOMAIN", + "PYTHON_PATH", + "DENO_PATH", + "GO_PATH", + "GOPRIVATE", + "GOPROXY", + "NETRC", + "PIP_INDEX_URL", + "PIP_EXTRA_INDEX_URL", + "PIP_TRUSTED_HOST", + "PATH", + "HOME", + "DATABASE_CONNECTIONS", + "TIMEOUT_WAIT_RESULT", + "QUEUE_LIMIT_WAIT_RESULT", + "DENO_AUTH_TOKENS", + "DENO_FLAGS", + "NPM_CONFIG_REGISTRY", + "PIP_LOCAL_DEPENDENCIES", + "ADDITIONAL_PYTHON_PATHS", + "INCLUDE_HEADERS", + "INSTANCE_EVENTS_WEBHOOK", + "CLOUD_HOSTED", + "GLOBAL_CACHE_INTERVAL", + "WORKER_TAGS", + "CUSTOM_TAGS", + "JOB_RETENTION_SECS", + "WAIT_RESULT_FAST_POLL_DURATION_SECS", + "WAIT_RESULT_SLOW_POLL_INTERVAL_MS", + "WAIT_RESULT_FAST_POLL_INTERVAL_MS", + "EXIT_AFTER_NO_JOB_FOR_SECS", + "REQUEST_SIZE_LIMIT", + "SMTP_HOST", + "SMTP_USERNAME", + "SMTP_PORT", + "SMTP_TLS_IMPLICIT", + "CREATE_WORKSPACE_REQUIRE_SUPERADMIN", + "GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE", +]; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 278d2f4fd4..af5b33441a 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -17,6 +17,7 @@ pub mod error; pub mod external_ip; pub mod flow_status; pub mod flows; +pub mod global_settings; pub mod jobs; pub mod more_serde; pub mod oauth2; diff --git a/backend/windmill-worker/src/global_cache.rs b/backend/windmill-worker/src/global_cache.rs index a7451f248f..75032f8cb2 100644 --- a/backend/windmill-worker/src/global_cache.rs +++ b/backend/windmill-worker/src/global_cache.rs @@ -6,6 +6,11 @@ use itertools::Itertools; use rand::Rng; #[cfg(feature = "enterprise")] use std::process::Stdio; +#[cfg(feature = "enterprise")] +use windmill_common::DB; + +#[cfg(feature = "enterprise")] +use windmill_common::global_settings::WORKER_S3_BUCKET_SYNC; #[cfg(feature = "enterprise")] use tokio::{process::Command, sync::mpsc::Sender, time::Instant}; @@ -219,6 +224,25 @@ pub async fn copy_cache_to_bucket(bucket: &str) -> error::Result<()> { Ok(()) } +#[cfg(feature = "enterprise")] +pub async fn worker_s3_bucket_sync_enabled(db: &DB) -> bool { + let q = sqlx::query!( + "SELECT value FROM global_settings WHERE name = $1", + WORKER_S3_BUCKET_SYNC + ) + .fetch_optional(db) + .await; + + if let Ok(q) = q { + let r = q.map(|x| x.value.as_bool().unwrap_or(true)).unwrap_or(true); + tracing::info!("Got global setting {WORKER_S3_BUCKET_SYNC}: {}", r); + r + } else { + tracing::info!("Failed to get global setting {WORKER_S3_BUCKET_SYNC}"); + false + } +} + #[cfg(feature = "enterprise")] pub async fn copy_cache_to_bucket_as_tar(bucket: &str) { tracing::info!("Copying cache to bucket {bucket} as tar"); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index bab689e7d6..576f179372 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -53,7 +53,7 @@ use windmill_queue::{add_completed_job, add_completed_job_error,IDLE_WORKERS}; use crate::{ worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, - }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs, write_file, transform_json_value, save_in_cache, hash_args}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::{handle_deno_job, generate_deno_lock}, + }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs, write_file, transform_json_value, save_in_cache, hash_args}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, graphql_executor::do_graphql, bun_executor::{handle_bun_job, gen_lockfile}, bash_executor::{ANSI_ESCAPE_RE, handle_powershell_job, handle_bash_job}, deno_executor::{handle_deno_job, generate_deno_lock}, global_cache::worker_s3_bucket_sync_enabled, }; #[cfg(feature = "enterprise")] @@ -442,16 +442,18 @@ pub async fn run_worker *GLOBAL_CACHE_INTERVAL && (copy_cache_from_bucket_handle.is_none() || copy_cache_from_bucket_handle.as_ref().unwrap().is_finished()) { diff --git a/frontend/src/routes/(root)/(logged)/workers/+page.svelte b/frontend/src/routes/(root)/(logged)/workers/+page.svelte index 9f260fd3c3..a6f53244c2 100644 --- a/frontend/src/routes/(root)/(logged)/workers/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workers/+page.svelte @@ -6,8 +6,10 @@ import Cell from '$lib/components/table/Cell.svelte' import DataTable from '$lib/components/table/DataTable.svelte' import Head from '$lib/components/table/Head.svelte' + import Toggle from '$lib/components/Toggle.svelte' import Tooltip from '$lib/components/Tooltip.svelte' - import { WorkerService, type WorkerPing } from '$lib/gen' + import { WorkerService, type WorkerPing, SettingService } from '$lib/gen' + import { enterpriseLicense } from '$lib/stores' import { sendUserToast } from '$lib/toast' import { displayDate, groupBy } from '$lib/utils' import { onDestroy, onMount } from 'svelte' @@ -17,6 +19,7 @@ let groupedWorkers: [string, WorkerPing[]][] = [] let intervalId: NodeJS.Timer | undefined + let globalCache = false $: filteredWorkers = (workers ?? []).filter((x) => (x.last_ping ?? 0) < 300) $: groupedWorkers = groupBy( filteredWorkers, @@ -24,6 +27,7 @@ (wp: WorkerPing) => wp.worker ) + const worker_s3_bucket_sync = 'worker_s3_bucket_sync' let timeSinceLastPing = 0 async function loadWorkers(): Promise { @@ -42,8 +46,17 @@ secondInterval = setInterval(() => { timeSinceLastPing += 1 }, 1000) + loadGlobalCache() }) + async function loadGlobalCache() { + try { + globalCache = (await SettingService.getGlobal({ key: worker_s3_bucket_sync })) ?? false + } catch (err) { + sendUserToast(`Could not load global cache: ${err}`, true) + } + } + onDestroy(() => { if (intervalId) { clearInterval(intervalId) @@ -62,6 +75,31 @@ documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups" /> +
+ global cache to s3 is an enterprise feature that enable workers to do fast cold start and + share a single cache backed by s3 to ensure that even with a high number of workers, + dependencies for python/deno/bun/go are only downloaded for the first time only once by the + whole fleet. require S3_CACHE_BUCKET to be set. + { + try { + console.log('Setting global cache to', e.detail) + await SettingService.setGlobal({ + key: worker_s3_bucket_sync, + requestBody: { value: e.detail } + }) + globalCache = e.detail + } catch (err) { + sendUserToast(`Could not set global cache: ${err}`, true) + } + }} + options={{ right: 'global cache to s3' }} + disabled={!$enterpriseLicense} + /> +
{#if workers != undefined} {#if groupedWorkers.length == 0}

No workers seems to be available