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

feat: autoscaling v0 (#4593)

Browse Source 
* all

* all

all

all

all

all

all

all

all

all

* all

* all

* nits

* nits
This commit is contained in:
Ruben Fiszel
2024-10-28 00:47:19 +01:00
committed by GitHub
parent bac3205725
commit fe7d044a66
29 changed files with 833 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
backend/.sqlx/query-02b516dac764662194db1bc33e365c01f40bae70af3683f1f09748f6020f0d49.json generated
View File

@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag, count(*) as count FROM queue WHERE\n scheduled_for <= now() - ('3 seconds')::interval AND running = false\n GROUP BY tag",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "count",
"type_info": "Int8"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null
]
},
"hash": "02b516dac764662194db1bc33e365c01f40bae70af3683f1f09748f6020f0d49"
}

28
backend/.sqlx/query-1f5f0858909eb5bac63c4e3b1add95226bd94ca3facb92a620ffa59dacad6705.json generated Normal file
View File

@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO autoscaling_event (worker_group, event_type, desired_workers, reason) VALUES ($1, $2, $3, $4)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
{
"Custom": {
"name": "autoscaling_event_type",
"kind": {
"Enum": [
"full_scaleout",
"scalein",
"scaleout"
]
}
}
},
"Int4",
"Text"
]
},
"nullable": []
},
"hash": "1f5f0858909eb5bac63c4e3b1add95226bd94ca3facb92a620ffa59dacad6705"
}

39
backend/.sqlx/query-4eca060026a0cb19c5794cd56ace89fc04765191f251945d14bbe78718714f6e.json generated Normal file
View File

@@ -0,0 +1,39 @@
{
"db_name": "PostgreSQL",
"query": "SELECT event_type::AUTOSCALING_EVENT_TYPE AS \"event_type: _\", EXTRACT(EPOCH FROM (NOW() - applied_at))::int as seconds_ago FROM autoscaling_event WHERE worker_group = $1 ORDER BY applied_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "event_type: _",
"type_info": {
"Custom": {
"name": "autoscaling_event_type",
"kind": {
"Enum": [
"full_scaleout",
"scalein",
"scaleout"
]
}
}
}
},
{
"ordinal": 1,
"name": "seconds_ago",
"type_info": "Int4"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
null
]
},
"hash": "4eca060026a0cb19c5794cd56ace89fc04765191f251945d14bbe78718714f6e"
}

52
backend/.sqlx/query-6d134b137ae81534e145fc5b6474cf963ee26a3ad3a0a3d8dc064cb14c8fd9a6.json generated Normal file
View File

@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, worker_group, event_type::text, desired_workers, reason, applied_at FROM autoscaling_event WHERE worker_group = $1 ORDER BY applied_at DESC LIMIT 5",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Int4"
},
{
"ordinal": 1,
"name": "worker_group",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "event_type",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "desired_workers",
"type_info": "Int4"
},
{
"ordinal": 4,
"name": "reason",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "applied_at",
"type_info": "Timestamp"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
null,
false,
true,
false
]
},
"hash": "6d134b137ae81534e145fc5b6474cf963ee26a3ad3a0a3d8dc064cb14c8fd9a6"
}

16
backend/Cargo.lock generated
View File

@@ -10426,6 +10426,7 @@ dependencies = [
"uuid 1.11.0",
"windmill-api",
"windmill-api-client",
"windmill-autoscaling",
"windmill-common",
"windmill-git-sync",
"windmill-indexer",
@@ -10549,6 +10550,21 @@ dependencies = [
"windmill-common",
]
[[package]]
name = "windmill-autoscaling"
version = "1.412.0"
dependencies = [
"anyhow",
"rsmq_async",
"serde",
"serde_json",
"sqlx",
"tracing",
"uuid 1.11.0",
"windmill-common",
"windmill-queue",
]
[[package]]
name = "windmill-common"
version = "1.412.0"

5
backend/Cargo.toml
View File

@@ -13,6 +13,7 @@ members = [
"./windmill-common",
"./windmill-audit",
"./windmill-git-sync",
"./windmill-autoscaling",
"./windmill-indexer",
"./windmill-macros",
"./parsers/windmill-parser",
@@ -45,7 +46,7 @@ lto = "thin"
[features]
default = []
enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise", "windmill-indexer/enterprise"]
enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise", "windmill-indexer/enterprise"]
enterprise_saml = ["windmill-api/enterprise_saml"]
stripe = ["windmill-api/stripe"]
benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark", "windmill-common/benchmark"]
@@ -72,6 +73,7 @@ windmill-git-sync.workspace = true
windmill-api = { workspace = true, default-features = false }
windmill-worker.workspace = true
windmill-indexer = { workspace = true, optional = true }
windmill-autoscaling = { workspace = true, optional = true }
futures.workspace = true
tracing.workspace = true
sqlx.workspace = true
@@ -116,6 +118,7 @@ windmill-worker = { path = "./windmill-worker" }
windmill-common = { path = "./windmill-common", default-features = false }
windmill-audit = { path = "./windmill-audit" }
windmill-git-sync = { path = "./windmill-git-sync" }
windmill-autoscaling = { path = "./windmill-autoscaling" }
windmill-indexer = {path = "./windmill-indexer"}
windmill-macros = {path = "./windmill-macros"}
windmill-parser = { path = "./parsers/windmill-parser" }

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

@@ -1 +1 @@
d61c163e0a311ecd86d1398aff883eaea8d0b09a
816abd18b24831f251e11d44d274c8acef89df12

3
backend/migrations/20241024125924_autoscaling.down.sql Normal file
View File

@@ -0,0 +1,3 @@
-- Add down migration script here
DROP TABLE autoscaling_event;
DROP TYPE autoscaling_event_type;

13
backend/migrations/20241024125924_autoscaling.up.sql Normal file
View File

@@ -0,0 +1,13 @@
-- Add up migration script here
CREATE TYPE AUTOSCALING_EVENT_TYPE AS ENUM ('full_scaleout', 'scalein', 'scaleout');
CREATE TABLE autoscaling_event (
id SERIAL PRIMARY KEY,
worker_group TEXT NOT NULL,
event_type AUTOSCALING_EVENT_TYPE NOT NULL,
desired_workers INTEGER NOT NULL,
applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
reason TEXT
);
CREATE INDEX autoscaling_event_worker_group_idx ON autoscaling_event (worker_group, applied_at);

7
backend/src/main.rs
View File

@@ -548,8 +548,11 @@ Windmill Community Edition {GIT_VERSION}
rx.recv().await?;
}
}
tracing::info!("Starting phase 2 of shutdown");
killpill_phase2_tx.send(())?;
if killpill_phase2_tx.receiver_count() > 0 {
tracing::info!("Starting phase 2 of shutdown");
killpill_phase2_tx.send(())?;
tracing::info!("Phase 2 of shutdown completed");
}
Ok(()) as anyhow::Result<()>
};

26
backend/src/monitor.rs
View File

@@ -27,7 +27,7 @@ use windmill_api::{
DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN,
};
#[cfg(feature = "enterprise")]
use windmill_common::ee::{worker_groups_alerts, jobs_waiting_alerts};
use windmill_common::ee::{jobs_waiting_alerts, worker_groups_alerts};
use windmill_common::{
auth::JWT_SECRET,
ee::CriticalErrorChannel,
@@ -1068,6 +1068,15 @@ pub async fn monitor_db(
}
};
let apply_autoscaling_f = async {
#[cfg(feature = "enterprise")]
if server_mode && !initial_load {
if let Err(e) = windmill_autoscaling::apply_all_autoscaling(db).await {
tracing::error!("Error applying autoscaling: {:?}", e);
}
}
};
join!(
expired_items_f,
zombie_jobs_f,
@@ -1075,6 +1084,7 @@ pub async fn monitor_db(
verify_license_key_f,
worker_groups_alerts_f,
jobs_waiting_alerts_f,
apply_autoscaling_f,
);
}
@@ -1092,19 +1102,11 @@ pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
.unwrap_or(true);
if metrics_enabled || save_metrics {
let queue_counts = sqlx::query!(
"SELECT tag, count(*) as count FROM queue WHERE
scheduled_for <= now() - ('3 seconds')::interval AND running = false
GROUP BY tag"
)
.fetch_all(db)
.await
.ok()
.unwrap_or_else(|| vec![]);
let queue_counts = windmill_common::queue::get_queue_counts(db).await;
for q in queue_counts {
let count = q.count.unwrap_or(0);
let tag = q.tag;
let count = q.1;
let tag = q.0;
if metrics_enabled {
let metric = (*QUEUE_COUNT).with_label_values(&[&tag]);
metric.set(count as i64);

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

@@ -8198,6 +8198,29 @@ paths:
items:
$ref: "#/components/schemas/Config"
/configs/list_autoscaling_events/{worker_group}:
get:
summary: List autoscaling events
operationId: listAutoscalingEvents
tags:
- config
parameters:
- name: worker_group
in: path
required: true
schema:
type: string
responses:
"200":
description: List of autoscaling events
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/AutoscalingEvent"
/w/{workspace}/acls/get/{kind}/{path}:
get:
summary: get granular acls
@@ -12469,3 +12492,21 @@ components:
properties:
dancer:
type: string
AutoscalingEvent:
type: object
properties:
id:
type: integer
format: int64
worker_group:
type: string
event_type:
type: string
desired_workers:
type: integer
reason:
type: string
applied_at:
type: string
format: date-time

28
backend/windmill-api/src/configs.rs
View File

@@ -29,6 +29,10 @@ pub fn global_service() -> Router {
.route("/update/:name", post(update_config).delete(delete_config))
.route("/get/:name", get(get_config))
.route("/list", get(list_configs))
.route(
"/list_autoscaling_events/:worker_group",
get(list_autoscaling_events),
)
}
#[derive(Serialize, Deserialize, FromRow)]
@@ -177,6 +181,30 @@ async fn delete_config(
Ok(format!("Deleted config {name}"))
}
#[derive(Serialize, Deserialize, FromRow)]
struct AutoscalingEvent {
id: i64,
worker_group: String,
event_type: Option<String>,
desired_workers: i32,
reason: Option<String>,
applied_at: chrono::NaiveDateTime,
}
async fn list_autoscaling_events(
Extension(db): Extension<DB>,
Path(worker_group): Path<String>,
) -> error::JsonResult<Vec<AutoscalingEvent>> {
let events = sqlx::query_as!(
AutoscalingEvent,
"SELECT id, worker_group, event_type::text, desired_workers, reason, applied_at FROM autoscaling_event WHERE worker_group = $1 ORDER BY applied_at DESC LIMIT 5",
worker_group
)
.fetch_all(&db)
.await?;
Ok(Json(events))
}
#[cfg(feature = "enterprise")]
async fn list_configs(
authed: ApiAuthed,

31
backend/windmill-api/src/jobs.rs
View File

@@ -2815,7 +2815,6 @@ pub async fn run_flow_by_path_inner(
let flow_path = flow_path.to_path();
check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
let (tag, dedicated_worker, has_preprocessor) = sqlx::query!(
"SELECT tag, dedicated_worker, flow_version.value->>'preprocessor_module' IS NOT NULL as has_preprocessor
FROM flow
@@ -2910,7 +2909,6 @@ pub async fn restart_flow(
) -> error::Result<(StatusCode, String)> {
check_license_key_valid().await?;
let completed_job = sqlx::query_as::<_, CompletedJob>(
"SELECT *, result->'wm_labels' as labels from completed_job WHERE id = $1 and workspace_id = $2",
)
@@ -3010,7 +3008,6 @@ pub async fn run_script_by_path_inner(
check_scopes(&authed, || format!("run:script/{script_path}"))?;
let (job_payload, tag, _delete_after_use, timeout) =
script_path_to_payload(script_path, &db, &w_id, run_query.skip_preprocessor).await?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
@@ -3018,7 +3015,6 @@ pub async fn run_script_by_path_inner(
let tag = run_query.tag.clone().or(tag);
check_tag_available_for_workspace(&w_id, &tag).await?;
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
let (uuid, tx) = push(
@@ -3067,7 +3063,6 @@ pub async fn run_workflow_as_code(
Query(wkflow_query): Query<WorkflowAsCodeQuery>,
Json(task): Json<WorkflowTask>,
) -> error::Result<(StatusCode, String)> {
let mut i = 1;
if *CLOUD_HOSTED {
@@ -3079,23 +3074,18 @@ pub async fn run_workflow_as_code(
check_license_key_valid().await?;
check_tag_available_for_workspace(&w_id, &run_query.tag).await?;
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
let job = get_queued_job(&job_id, &w_id, &db).await?;
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
let job = not_found_if_none(job, "Queued Job", &job_id.to_string())?;
let (job_payload, tag, _delete_after_use, timeout) = match job.job_kind {
JobKind::Preview => (
@@ -3118,18 +3108,12 @@ pub async fn run_workflow_as_code(
run_query.timeout,
),
JobKind::Script => {
script_path_to_payload(
job.script_path(),
&db,
&w_id,
run_query.skip_preprocessor,
)
.await?
script_path_to_payload(job.script_path(), &db, &w_id, run_query.skip_preprocessor)
.await?
}
_ => return Err(anyhow::anyhow!("Not supported").into()),
};
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
@@ -3143,22 +3127,18 @@ pub async fn run_workflow_as_code(
let tag = run_query.tag.clone().or(tag).or(Some(job.tag));
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
let tx = PushIsolationLevel::Isolated(user_db, authed.clone().into(), rsmq);
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
let (uuid, mut tx) = push(
&db,
tx,
@@ -3185,7 +3165,6 @@ pub async fn run_workflow_as_code(
)
.await?;
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
@@ -3203,16 +3182,13 @@ pub async fn run_workflow_as_code(
tracing::info!("Skipping update of flow status for job {job_id} in workspace {w_id}");
}
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
i += 1;
}
tx.commit().await?;
if *CLOUD_HOSTED {
tracing::info!("workflow_as_code_tracing id {i} ");
}
@@ -3866,7 +3842,6 @@ pub async fn run_wait_result_flow_by_path_internal(
let flow_path = flow_path.to_path();
check_scopes(&authed, || format!("run:flow/{flow_path}"))?;
let scheduled_for = run_query.get_scheduled_for(&db).await?;
let (tag, dedicated_worker, early_return, has_preprocessor) = sqlx::query!(
@@ -4362,7 +4337,6 @@ async fn add_batch_jobs(
}
}
"flow" => {
let mut uuids: Vec<Uuid> = Vec::new();
let payload = if let Some(ref fv) = batch_info.flow_value {
JobPayload::RawFlow { value: fv.clone(), path: None, restarted_from: None }
@@ -4580,7 +4554,6 @@ pub async fn run_job_by_hash_inner(
#[cfg(feature = "enterprise")]
check_license_key_valid().await?;
let hash = script_hash.0;
let (
path,

24
backend/windmill-autoscaling/Cargo.toml Normal file
View File

@@ -0,0 +1,24 @@
[package]
name = "windmill-autoscaling"
version.workspace = true
authors.workspace = true
edition.workspace = true
[lib]
name = "windmill_autoscaling"
path = "./src/lib.rs"
[features]
enterprise = ["windmill-queue/enterprise", "windmill-common/enterprise"]
default = []
[dependencies]
uuid.workspace = true
serde.workspace = true
sqlx.workspace = true
serde_json.workspace = true
tracing.workspace = true
windmill-common = { workspace = true, default-features = false }
windmill-queue.workspace = true
rsmq_async.workspace = true
anyhow.workspace = true

6
backend/windmill-autoscaling/src/autoscaling_ee.rs Normal file
View File

@@ -0,0 +1,6 @@
use windmill_common::DB;
pub async fn apply_all_autoscaling(_db: &DB) -> anyhow::Result<()> {
// Autoscaling is an ee feature
Ok(())
}

2
backend/windmill-autoscaling/src/lib.rs Normal file
View File

@@ -0,0 +1,2 @@
mod autoscaling_ee;
pub use autoscaling_ee::*;

4
backend/windmill-common/src/lib.rs
View File

@@ -17,6 +17,7 @@ use scripts::ScriptLang;
use sqlx::{Pool, Postgres};
pub mod apps;
pub mod auth;
#[cfg(feature = "benchmark")]
pub mod bench;
pub mod db;
@@ -32,9 +33,8 @@ pub mod job_s3_helpers_ee;
pub mod jobs;
pub mod more_serde;
pub mod oauth2;
pub mod queue;
pub mod s3_helpers;
pub mod auth;
pub mod schedule;
pub mod scripts;
pub mod server;

16
backend/windmill-common/src/queue.rs Normal file
View File

@@ -0,0 +1,16 @@
use std::collections::HashMap;
use sqlx::{Pool, Postgres};
pub async fn get_queue_counts(db: &Pool<Postgres>) -> HashMap<String, u32> {
sqlx::query_as::<_, (String, i64)>(
"SELECT tag, count(*) as count FROM queue WHERE
scheduled_for <= now() - ('3 seconds')::interval AND running = false
GROUP BY tag",
)
.fetch_all(db)
.await
.ok()
.map(|v| v.into_iter().map(|(k, v)| (k, v as u32)).collect())
.unwrap_or_else(|| HashMap::new())
}

2
backend/windmill-worker/src/pg_executor.rs
View File

@@ -85,7 +85,7 @@ fn do_postgresql_inner<'a>(
let arg_t = arg
.otyp
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Missing otyp for pg arg"))?;
.ok_or_else(|| anyhow::anyhow!("Missing otzyp for pg arg"))?;
let typ = &arg.typ;
let param = convert_val(value, arg_t, typ)?;
query_params.push(param);

364
frontend/src/lib/components/AutoscalingConfigEditor.svelte Normal file
View File

@@ -0,0 +1,364 @@
<script lang="ts">
import type { AutoscalingConfig } from './worker_group'
import Toggle from './Toggle.svelte'
import Section from './Section.svelte'
import Tooltip from './Tooltip.svelte'
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
import { Button } from './common'
import { ExternalLink } from 'lucide-svelte'
import { createEventDispatcher } from 'svelte'
import Label from './Label.svelte'
import MultiSelect from 'svelte-multiselect'
export let config: AutoscalingConfig | undefined
export let worker_tags: string[] | undefined
const dispatch = createEventDispatcher()
let test_input: number = 3
</script>
<div class="flex flex-row gap-16 pt-2">
<div class="space-y-4 flex flex-col gap-1 max-w-xs text-sm">
<h5>Rules</h5>
<Toggle
checked={config?.enabled ?? false}
options={{ right: 'Enabled' }}
on:change={(e) => {
dispatch('dirty')
if (e.detail) {
if (!config) {
config = {
enabled: true,
min_workers: 3,
max_workers: 10,
integration: { type: 'dryrun' }
}
} else {
config.enabled = true
}
} else {
config = {
...(config ?? {
min_workers: 3,
max_workers: 10,
integration: { type: 'dryrun' }
}),
enabled: false
}
}
}}
/>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Min # of Workers
{#if config !== undefined}
<input on:input={() => dispatch('dirty')} type="number" bind:value={config.min_workers} />
{#if config.min_workers !== undefined && config.min_workers != undefined && config.min_workers > config.max_workers}
<div class="text-red-600 text-xs whitespace-nowrap"
>Minimum cannot be {'>'} to Maximum</div
>
{/if}
{:else}
<input type="number" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Max # of Workers
{#if config !== undefined}
<input on:input={() => dispatch('dirty')} type="number" bind:value={config.max_workers} />
{:else}
<input type="number" disabled />
{/if}
</label>
<div class="p-2">
<Section label="Advanced" small collapsable={true}>
<div class="flex flex-col gap-2 text-2xs">
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Cooldown seconds after an incremental scale-in/out
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
step="1"
min="30"
placeholder="300"
bind:value={config.cooldown_seconds}
/>
{:else}
<input type="number" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Cooldown seconds after a full scale out
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
step="1"
min="30"
placeholder="1500"
bind:value={config.full_scale_cooldown_seconds}
/>
{:else}
<input type="number" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Num jobs waiting to trigger an incremental scale-out
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
bind:value={config.inc_scale_num_jobs_waiting}
placeholder="1"
/>
{:else}
<input type="number" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Num jobs waiting to trigger a full scale out <Tooltip
>Default: max_workers, full scale out = scale out to max workers</Tooltip
>
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
placeholder="max workers"
bind:value={config.full_scale_jobs_waiting}
/>
{:else}
<input type="number" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Occupancy rate % threhsold to go below to trigger a scale-in (decrease) <Tooltip
>Default: 25%, need to go below average of all of 15s, 5m and 30m occupancy rates</Tooltip
>
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
step="1"
min="0"
max="100"
placeholder="25"
bind:value={config.dec_scale_occupancy_rate}
/>
{:else}
<input type="number" step="0.01" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Occupancy rate threshold to exceed to trigger an incremental scale-out (increase) <Tooltip
>Default: 75%, need to exceed average of all of 15s, 5m and 30m occupancy rates</Tooltip
>
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
step="1"
min="0"
max="100"
placeholder="75"
bind:value={config.inc_scale_occupancy_rate}
/>
{:else}
<input type="number" step="0.01" disabled />
{/if}
</label>
<!-- svelte-ignore a11y-label-has-associated-control -->
<label>
Num workers to scale-in/out by when incremental <Tooltip
>Default: (max_workers - min_workers) / 5</Tooltip
>
{#if config !== undefined}
<input
on:input={() => dispatch('dirty')}
type="number"
step="1"
min="1"
placeholder="(max_workers - min_workers) / 5"
bind:value={config.inc_num_workers}
/>
{:else}
<input type="number" disabled />
{/if}
</label>
<Label label="Custom tags to autoscale on">
<svelte:fragment slot="header">
<Tooltip>
By default, autoscaling will apply to the tags the worker group is assigned to but
you can override this here.
</Tooltip>
</svelte:fragment>
{#if config}
{#if config.custom_tags}
<MultiSelect
outerDivClass="text-secondary !bg-surface-disabled !border-0"
selected={config.custom_tags}
on:change={(e) => {
console.log(e.detail.type, config?.custom_tags)
if (e.detail && config?.custom_tags) {
if (e.detail.type === 'add') {
config.custom_tags = [...config.custom_tags, e.detail.option]
} else if (e.detail.type === 'remove') {
config.custom_tags = config.custom_tags.filter((t) => t !== e.detail.option)
if (config?.custom_tags && config.custom_tags.length == 0) {
config.custom_tags = undefined
}
} else if (e.detail.type === 'removeAll') {
config.custom_tags = undefined
} else {
console.error(
`Priority tags multiselect - unknown event type: '${e.detail.type}'`
)
}
dispatch('dirty')
}
}}
on:clear={() => {
if (config) {
config.custom_tags = undefined
}
dispatch('dirty')
}}
options={worker_tags ?? []}
selectedOptionsDraggable={false}
ulOptionsClass={'!bg-surface-secondary'}
placeholder="Tags"
/>
{:else}
<Button
color="light"
size="xs"
variant="contained"
on:click={() => {
if (config) {
config.custom_tags = []
dispatch('dirty')
}
}}>Add custom tags</Button
>
{/if}
{/if}
</Label>
</div>
</Section>
</div>
</div>
<div class="flex flex-col gap-1 max-w-xs text-sm">
<h5>Integration</h5>
{#if config?.integration}
<ToggleButtonGroup
on:selected={(e) => dispatch('dirty')}
bind:selected={config.integration.type}
class="mb-4 mt-2"
>
<ToggleButton
value="dryrun"
size="sm"
label="Dry run"
tooltip="See autoscaling events but not actual scaling actions will be performed"
/>
<ToggleButton
value="script"
size="sm"
label="Custom Script"
tooltip="Run a custom script to scale your worker group"
/>
<ToggleButton position="center" disabled value="ecs" size="sm" label="ECS (soon)" />
<ToggleButton position="right" disabled value="nomad" size="sm" label="Nomad (soon)" />
<ToggleButton
position="right"
disabled
value="kubernetes"
size="sm"
label="Kubernetes (soon)"
/>
</ToggleButtonGroup>
{#if config.integration.type === 'script'}
<label>
Script path on the 'admins' workspace
<input
on:input={() => dispatch('dirty')}
type="text"
bind:value={config.integration.path}
/>
</label>
<label>
Custom tag for executing script (optional)
{#if config.integration.tag}
<input
on:input={() => dispatch('dirty')}
type="text"
bind:value={config.integration.tag}
/>
{:else}
<Button
color="light"
size="xs"
variant="contained"
on:click={() => {
if (config?.integration?.type === 'script') {
config.integration.tag = 'bash'
dispatch('dirty')
}
}}>Set tag</Button
>
{/if}
</label>
<div class="flex mt-6 gap-2">
<Button
color="dark"
target="_blank"
endIcon={{ icon: ExternalLink }}
href="/scripts/add?hub=hub%2F9204%2Fhelper%2FScale%20a%20worker%20group%20deployed%20as%20a%20kubernetes%20service&workspace=admins"
>Create from template</Button
>
<Button
color="dark"
target="_blank"
href={`/runs/${config.integration.path}?workspace=admins`}
endIcon={{ icon: ExternalLink }}
>
See jobs
</Button>
</div>
<div class="flex flex-row gap-2 mt-4">
<Button color="light" size="xs" variant="contained">Test scaling</Button>
<div class="flex text-xs flex-row gap-2 items-center">
<input class="!w-16" type="number" bind:value={test_input} />
workers
</div>
</div>
{/if}
{:else}
<ToggleButtonGroup selected={'script'} disabled class="mb-4 mt-2">
<ToggleButton value="dryrun" size="sm" label="Dry run" />
<ToggleButton value="script" size="sm" label="Custom Script" />
<ToggleButton position="center" value="ecs" size="sm" label="ECS (soon)" />
<ToggleButton position="right" value="nomad" size="sm" label="Nomad (soon)" />
<ToggleButton position="right" value="kubernetes" size="sm" label="Kubernetes (soon)" />
</ToggleButtonGroup>
<label>
Script path on the 'admins' workspace
<input type="text" disabled />
</label>
{/if}
</div>
</div>

73
frontend/src/lib/components/AutoscalingEvents.svelte Normal file
View File

@@ -0,0 +1,73 @@
<script lang="ts">
import { ConfigService, type AutoscalingEvent } from '$lib/gen'
import { LoaderIcon, RefreshCw } from 'lucide-svelte'
import { Button, Skeleton } from './common'
import { twMerge } from 'tailwind-merge'
import TimeAgo from './TimeAgo.svelte'
import { enterpriseLicense } from '$lib/stores'
export let worker_group: string
let loading = true
let events: AutoscalingEvent[] | undefined = undefined
$: worker_group && loadEvents()
async function loadEvents() {
loading = true
try {
events = await ConfigService.listAutoscalingEvents({ workerGroup: worker_group })
} catch (e) {
events = []
console.error(e)
} finally {
loading = false
}
}
</script>
<div>
<h6
class={!$enterpriseLicense || (events != undefined && events.length == 0)
? 'text-tertiary'
: ''}
>Autoscaling events {#if $enterpriseLicense}<span class="text-xs text-tertiary">(5 last)</span>
<span class="inline-flex ml-6">
<Button
startIcon={{
icon: loading ? LoaderIcon : RefreshCw,
classes: twMerge(
loading ? 'animate-spin text-blue-800' : '',
'transition-all text-gray-500 dark:text-white'
)
}}
color="light"
size="xs2"
btnClasses={twMerge(loading ? ' bg-blue-100 dark:bg-blue-400' : '', 'transition-all')}
on:click={() => loadEvents()}
iconOnly
/>
</span>{/if}
</h6>
{#if !$enterpriseLicense}
<div class="text-xs pt-2 text-tertiary">Autoscaling is an EE feature</div>
{:else if loading}
<Skeleton layout={[[12], 1]} />
{:else if events}
{#if events.length == 0}
<div class="text-xs pt-2 text-tertiary"
>No events, is autoscaling set in the worker group config?</div
>
{:else}
<div class="flex flex-col gap-2 text-xs text-tertiary pt-4">
{#each events as event}
<div class="flex flex-row gap-4">
<div class="text-primary">{event.event_type} to {event.desired_workers}</div>
<div class="text-secondary">{event.reason}</div>
<div class="text-tertiary"><TimeAgo date={event.applied_at ?? ''} /></div>
</div>
{/each}
</div>
{/if}
{/if}
</div>

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

@@ -277,7 +277,9 @@
<div class="text-secondary pb-4 text-xs"
>Setting SMTP unlocks sending emails upon adding new users to the workspace or the
instance or sending critical alerts.
<a target="_blank" href="https://www.windmill.dev/docs/advanced/instance_settings#smtp">Learn more</a
<a
target="_blank"
href="https://www.windmill.dev/docs/advanced/instance_settings#smtp">Learn more</a
></div
>
{:else if category == 'Registries'}

2
frontend/src/lib/components/OAuthSetting.svelte
View File

@@ -66,7 +66,7 @@
>
{#if enabled}
<div class="p-2 rounded border">
{#if name != "slack"}
{#if name != 'slack'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Custom Name</span>
<input type="text" placeholder="Custom Name" bind:value={value['display_name']} />

26
frontend/src/lib/components/WorkspaceGroup.svelte → frontend/src/lib/components/WorkerGroup.svelte
View File

@@ -1,6 +1,6 @@
<script lang="ts">
import { Copy, Plus, RefreshCcwIcon, Settings, Trash, X } from 'lucide-svelte'
import { Alert, Button, Drawer } from './common'
import { Alert, Badge, Button, Drawer } from './common'
import Multiselect from 'svelte-multiselect'
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
@@ -18,6 +18,8 @@
import AutoComplete from 'simple-svelte-autocomplete'
import YAML from 'yaml'
import Toggle from './Toggle.svelte'
import type { AutoscalingConfig } from './worker_group'
import AutoscalingConfigEditor from './AutoscalingConfigEditor.svelte'
export let name: string
export let config:
@@ -31,6 +33,7 @@
additional_python_paths?: string[]
pip_local_dependencies?: string[]
min_alive_workers_alert_threshold?: number
autoscaling?: AutoscalingConfig
}
export let activeWorkers: number
export let customTags: string[] | undefined
@@ -65,6 +68,7 @@
additional_python_paths?: string[]
pip_local_dependencies?: string[]
min_alive_workers_alert_threshold?: number
autoscaling?: AutoscalingConfig
} = {}
function loadNConfig() {
@@ -438,7 +442,7 @@
</Tooltip>
</svelte:fragment>
<Multiselect
outerDivClass="text-secondary !bg-surface-disabled"
outerDivClass="text-secondary !bg-surface-disabled !border-0"
disabled={!$enterpriseLicense}
bind:selected={selectedPriorityTags}
on:change={(e) => {
@@ -452,6 +456,9 @@
delete nconfig.priority_tags[e.detail.option]
}
dirty = true
} else if (e.detail.type === 'removeAll') {
nconfig.priority_tags = undefined
dirty = true
} else {
console.error(
`Priority tags multiselect - unknown event type: '${e.detail.type}'`
@@ -782,6 +789,21 @@
</Section>
<div class="mt-8" />
<Section label="Autoscaling" collapsable>
<div slot="header" class="ml-4 flex flex-row gap-2 items-center">
<Badge>Beta</Badge>
{#if nconfig.autoscaling?.enabled}
<Badge color="green">Enabled</Badge>
{/if}
</div>
<AutoscalingConfigEditor
on:dirty={() => (dirty = true)}
worker_tags={nconfig.worker_tags}
bind:config={nconfig.autoscaling}
/>
</Section>
<div class="mt-8" />
<Section
label="Init Script"
tooltip="Bash scripts run at start of the workers. More lightweight than having to require the worker images at the cost of being run on every start."

2
frontend/src/lib/components/flows/content/FlowSettings.svelte
View File

@@ -441,7 +441,7 @@
}}
options={{
right: `Label as high priority`,
rightTooltip: `All jobs scheduled by flows labeled as high priority take precedence over the other jobs in the jobs queue. ${
rightTooltip: `All jobs scheduled by flows labeled as high priority take precedence over the other jobs in the jobs queue. Higher priority numbers are executed first. ${
!$enterpriseLicense ? 'This is a feature only available on enterprise edition.' : ''
}`
}}

39
frontend/src/lib/components/worker_group.ts Normal file
View File

@@ -0,0 +1,39 @@
// enabled: bool,
// min_workers: usize,
// max_workers: usize,
// cooldown_seconds: usize,
// inc_scale_jobs_waiting: usize,
// full_scale_cooldown_seconds: usize,
// full_scale_jobs_waiting: usize,
// dec_scale_occupancy_rate: u8, // occupancy rate of 30s, 5m, 30m to scale down
// inc_scale_occupancy_rate: u8, // occupancy rate of 30s, 5m to scale up
// inc_percent: usize,
// integration: Option<AutoscalingIntegration>,
export type AutoscalingConfig = {
enabled: boolean
min_workers: number
max_workers: number
cooldown_seconds?: number
inc_scale_num_jobs_waiting?: number
full_scale_cooldown_seconds?: number
full_scale_jobs_waiting?: number
dec_scale_occupancy_rate?: number
inc_scale_occupancy_rate?: number
inc_num_workers?: number
integration?: AutoscalingIntegration
custom_tags?: string[]
}
export type AutoscalingIntegration = AutoscaleScript | AutoscaleDryRun
export type AutoscaleScript = {
type: 'script'
tag?: string
path: string
}
export type AutoscaleDryRun = {
type: 'dryrun'
}

6
frontend/src/lib/utils.ts
View File

@@ -472,7 +472,7 @@ export async function copyToClipboard(value?: string, sendToast = true): Promise
}
export function pluralize(quantity: number, word: string, customPlural?: string) {
if (quantity <= 1) {
if (quantity == 1) {
return `${quantity} ${word}`
} else if (customPlural) {
return `${quantity} ${customPlural}}`
@@ -498,7 +498,7 @@ export function isObject(obj: any) {
export function debounce(func: (...args: any[]) => any, wait: number) {
let timeout: any
return function(...args: any[]) {
return function (...args: any[]) {
// @ts-ignore
const context = this
clearTimeout(timeout)
@@ -508,7 +508,7 @@ export function debounce(func: (...args: any[]) => any, wait: number) {
export function throttle<T>(func: (...args: any[]) => T, wait: number) {
let timeout: any
return function(...args: any[]) {
return function (...args: any[]) {
if (!timeout) {
timeout = setTimeout(() => {
timeout = null

52
frontend/src/routes/(root)/(logged)/workers/+page.svelte
View File

@@ -13,17 +13,18 @@
import DataTable from '$lib/components/table/DataTable.svelte'
import Head from '$lib/components/table/Head.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import WorkspaceGroup from '$lib/components/WorkspaceGroup.svelte'
import WorkspaceGroup from '$lib/components/WorkerGroup.svelte'
import { WorkerService, type WorkerPing, ConfigService, SettingService } from '$lib/gen'
import { enterpriseLicense, superadmin } from '$lib/stores'
import { sendUserToast } from '$lib/toast'
import { displayDate, groupBy, pluralize, truncate } from '$lib/utils'
import { AlertTriangle, CopyIcon, LineChart, List, Plus, Search } from 'lucide-svelte'
import { AlertTriangle, LineChart, List, Plus, Search } from 'lucide-svelte'
import { getContext, onDestroy, onMount } from 'svelte'
import AutoComplete from 'simple-svelte-autocomplete'
import YAML from 'yaml'
import { DEFAULT_TAGS_WORKSPACES_SETTING } from '$lib/consts'
import AutoscalingEvents from '$lib/components/AutoscalingEvents.svelte'
let workers: WorkerPing[] | undefined = undefined
let workerGroups: Record<string, any> | undefined = undefined
@@ -326,7 +327,7 @@
<p>No workers seem to be available</p>
{/if}
<div class="py-4 w-full flex justify-between"
<div class="pt-4 pb-8 w-full flex justify-between items-center"
><h4
>{groupWorkers?.length} Worker Groups <Tooltip
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups"
@@ -336,6 +337,7 @@
</Tooltip></h4
>
<div />
{#if $superadmin}
<div class="flex flex-row items-center">
<Popup
@@ -344,31 +346,30 @@
>
<svelte:fragment slot="button">
<div class="flex items-center gap-2">
<Button
size="sm"
color="light"
startIcon={{ icon: CopyIcon }}
on:click={() => {
if (!workerGroups) {
return sendUserToast('No worker groups found', true)
}
const workersConfig = Object.entries(workerGroups).map(([name, config]) => ({
name,
...config
}))
navigator.clipboard.writeText(YAML.stringify(workersConfig))
sendUserToast('Worker groups config copied to clipboard as YAML')
}}
>
<span class="hidden md:block">Copy groups config</span>
</Button>
<Button
size="sm"
startIcon={{ icon: Plus }}
nonCaptureEvent
disabled={!$enterpriseLicense}
dropdownItems={$enterpriseLicense
? [
{
label: 'Copy groups config as YAML',
onClick: () => {
if (!workerGroups) {
return sendUserToast('No worker groups found', true)
}
const workersConfig = Object.entries(workerGroups).map(
([name, config]) => ({
name,
...config
})
)
navigator.clipboard.writeText(YAML.stringify(workersConfig))
sendUserToast('Worker groups config copied to clipboard as YAML')
}
},
{
label: 'Import groups config from YAML',
onClick: () => {
@@ -378,7 +379,9 @@
]
: undefined}
>
<span class="hidden md:block">New group config</span>
<span class="hidden md:block"
>New group config {!$enterpriseLicense ? '(EE)' : ''}</span
>
<Tooltip light>
Worker Group configs are propagated to every workers in the worker group
@@ -644,7 +647,8 @@
{/if}
{/if}
</div>
<div class="pb-4" />
<div class="pb-20" />
<AutoscalingEvents worker_group={selectedTab} />
{:else}
<div class="flex flex-col">
{#each new Array(4) as _}