diff --git a/backend/.sqlx/query-8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80.json b/backend/.sqlx/query-8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80.json new file mode 100644 index 0000000000..be9f08517a --- /dev/null +++ b/backend/.sqlx/query-8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT timestamp AS \"timestamp!\",\n operation::text AS \"operation!\",\n resource AS workspace_id,\n parameters\n FROM (\n SELECT timestamp, operation, resource, parameters\n FROM audit_partitioned\n WHERE workspace_id = 'admins'\n AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')\n UNION ALL\n SELECT timestamp, operation, resource, parameters\n FROM audit\n WHERE workspace_id = 'admins'\n AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped')\n ) e\n ORDER BY timestamp DESC\n LIMIT 100\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "timestamp!", + "type_info": "Timestamptz" + }, + { + "ordinal": 1, + "name": "operation!", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "parameters", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "8df7878c46d8bfd38d4df91dff578e5d2d13f3516d954610b32b0709d6768a80" +} diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index dc6b08db08..4605315f3c 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -20,7 +20,10 @@ use windmill_common::{ error::JsonResult, jobs::{HIDE_WORKERS_FOR_NON_ADMINS, TAGS_ARE_SENSITIVE}, utils::{paginate, Pagination}, - worker::{ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE}, + worker::{ + is_cloud_production_host, ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, + DEFAULT_TAGS_PER_WORKSPACE, + }, DB, }; @@ -39,6 +42,10 @@ pub fn global_service() -> Router { .route("/queue_metrics", get(get_queue_metrics)) .route("/queue_counts", get(get_queue_counts)) .route("/queue_running_counts", get(get_queue_running_counts)) + .route( + "/workspace_fairness_events", + get(get_workspace_fairness_events), + ) } pub fn workspaced_service() -> Router { @@ -283,3 +290,58 @@ async fn get_queue_running_counts( let queue_running_counts = windmill_common::queue::get_queue_running_counts(&db).await; Ok(Json(queue_running_counts)) } + +#[derive(Serialize)] +pub struct WorkspaceFairnessEvent { + pub timestamp: chrono::DateTime, + pub operation: String, + /// Affected workspace (stored in audit log `resource`). `None` only for very + /// old rows pre-dating the resource convention — UI should treat as "unknown". + pub workspace_id: Option, + /// Snapshot of the relevant fairness settings at the time of the transition + /// (`max_percent`, `window_secs`, `total_overloaded`). `None` for uncap rows. + pub parameters: Option, +} + +/// Return the last 100 workspace-fairness cap/uncap transitions written by +/// `workspace_fairness::emit_transition_audit`. Cloud-only: the underlying +/// mechanism is hard-gated to `CLOUD_HOSTED=true` + `app.windmill.dev`, so on +/// any other instance we short-circuit with an empty list rather than running +/// a query that would always be empty. +async fn get_workspace_fairness_events( + authed: ApiAuthed, + Extension(db): Extension, +) -> JsonResult> { + require_devops_role(&db, &authed.email).await?; + + if !is_cloud_production_host() { + return Ok(Json(vec![])); + } + + let events = sqlx::query_as!( + WorkspaceFairnessEvent, + r#" + SELECT timestamp AS "timestamp!", + operation::text AS "operation!", + resource AS workspace_id, + parameters + FROM ( + SELECT timestamp, operation, resource, parameters + FROM audit_partitioned + WHERE workspace_id = 'admins' + AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped') + UNION ALL + SELECT timestamp, operation, resource, parameters + FROM audit + WHERE workspace_id = 'admins' + AND operation IN ('workspace_fairness.capped', 'workspace_fairness.uncapped') + ) e + ORDER BY timestamp DESC + LIMIT 100 + "#, + ) + .fetch_all(&db) + .await?; + + Ok(Json(events)) +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a72a6e31dc..64b90fa782 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -17682,6 +17682,38 @@ paths: additionalProperties: type: integer + /workers/workspace_fairness_events: + get: + summary: list last 100 workspace-fairness cap/uncap events (cloud-only) + operationId: getWorkspaceFairnessEvents + tags: + - worker + responses: + "200": + description: workspace fairness events (empty on non-cloud) + content: + application/json: + schema: + type: array + items: + type: object + properties: + timestamp: + type: string + format: date-time + operation: + type: string + workspace_id: + type: string + nullable: true + parameters: + type: object + nullable: true + additionalProperties: true + required: + - timestamp + - operation + /configs/list_worker_groups: get: summary: list worker groups diff --git a/backend/windmill-queue/src/workspace_fairness.rs b/backend/windmill-queue/src/workspace_fairness.rs index a5b39055a8..e4257d8d4c 100644 --- a/backend/windmill-queue/src/workspace_fairness.rs +++ b/backend/windmill-queue/src/workspace_fairness.rs @@ -37,12 +37,16 @@ //! cloud's `global_settings` row). When either check fails, [`maybe_refresh_overloaded`] //! and the pull-side dispatch both treat the feature as disabled. +use std::collections::HashSet; use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; use sqlx::{Pool, Postgres}; +use windmill_audit::audit_oss::audit_log; +use windmill_audit::ActionKind; +use windmill_common::audit::AuditAuthor; use windmill_common::error::Result; use windmill_common::worker::{ is_cloud_production_host, WORKSPACE_FAIRNESS_DURATION_SECS, WORKSPACE_FAIRNESS_ENABLED, @@ -190,6 +194,28 @@ async fn refresh_overloaded(db: &Pool) -> Result<()> { .await? .is_some(); + // Winner-only: read the value the cluster currently holds, so we can emit + // audit-log entries for workspaces that just entered or left the capped set. + // We compare against the DB-stored prior list (not the per-process local + // cache) so a freshly-restarted worker that happens to win the claim does + // not emit spurious "newly capped" entries for the workspaces that were + // already capped before it started. + let prev_list_from_db: Vec = if won { + let prev_val: Option = + sqlx::query_scalar("SELECT value FROM background_task_state WHERE name = $1") + .bind(TASK_STATE_NAME) + .fetch_optional(db) + .await?; + match prev_val { + Some(v) => serde_json::from_value::(v) + .map(|s| s.overloaded) + .unwrap_or_default(), + None => vec![], + } + } else { + vec![] + }; + // Step 2: winner-only aggregation + value write. `SET` is evaluated per // updated row, so issuing this statement only when `won` guarantees the // expensive aggregation never runs for a loser. @@ -247,6 +273,33 @@ async fn refresh_overloaded(db: &Pool) -> Result<()> { None => vec![], }; + // Audit-log transitions (winner only). Skipping losers prevents N duplicate + // entries per transition on a fleet of N worker processes. + if won && prev_list_from_db != new_list { + let prev_set: HashSet<&str> = prev_list_from_db.iter().map(String::as_str).collect(); + let new_set: HashSet<&str> = new_list.iter().map(String::as_str).collect(); + let newly_capped: Vec<&str> = new_list + .iter() + .map(String::as_str) + .filter(|w| !prev_set.contains(w)) + .collect(); + let newly_uncapped: Vec<&str> = prev_list_from_db + .iter() + .map(String::as_str) + .filter(|w| !new_set.contains(w)) + .collect(); + + emit_transition_audit( + db, + &newly_capped, + &newly_uncapped, + max_percent, + duration_secs, + new_list.len(), + ) + .await; + } + let prev = WORKSPACE_FAIRNESS_OVERLOADED.load(); if **prev != new_list { tracing::info!( @@ -260,3 +313,72 @@ async fn refresh_overloaded(db: &Pool) -> Result<()> { Ok(()) } + +/// Best-effort audit-log emission for workspaces entering or leaving the +/// capped set. Failures are logged but never propagated — the refresh cycle +/// must not abort just because an audit insert fails. +/// +/// Entries are scoped to the `admins` workspace and carry the affected +/// workspace as the `resource` field, so the queue-metrics drawer (cloud-only) +/// can fetch a single time-ordered stream of cap transitions without merging +/// across workspaces. Cluster admins also see them in the standard `admins` +/// audit listing (no `all_workspaces=true` toggle required). +async fn emit_transition_audit( + db: &Pool, + newly_capped: &[&str], + newly_uncapped: &[&str], + max_percent: i64, + duration_secs: i32, + total_overloaded: usize, +) { + if newly_capped.is_empty() && newly_uncapped.is_empty() { + return; + } + + let author = AuditAuthor { + username: "system".to_string(), + email: "system@windmill.dev".to_string(), + username_override: None, + token_prefix: None, + }; + + let max_percent_s = max_percent.to_string(); + let window_secs_s = duration_secs.to_string(); + let total_s = total_overloaded.to_string(); + + for w in newly_capped { + let mut params = std::collections::HashMap::<&str, &str>::new(); + params.insert("max_percent", &max_percent_s); + params.insert("window_secs", &window_secs_s); + params.insert("total_overloaded", &total_s); + if let Err(e) = audit_log( + db, + &author, + "workspace_fairness.capped", + ActionKind::Update, + "admins", + Some(w), + Some(params), + ) + .await + { + tracing::warn!("failed to write workspace_fairness.capped audit for {w}: {e:#}"); + } + } + + for w in newly_uncapped { + if let Err(e) = audit_log( + db, + &author, + "workspace_fairness.uncapped", + ActionKind::Update, + "admins", + Some(w), + None, + ) + .await + { + tracing::warn!("failed to write workspace_fairness.uncapped audit for {w}: {e:#}"); + } + } +} diff --git a/frontend/src/lib/components/QueueMetricsDrawer.svelte b/frontend/src/lib/components/QueueMetricsDrawer.svelte index 314a2985a7..617352ea27 100644 --- a/frontend/src/lib/components/QueueMetricsDrawer.svelte +++ b/frontend/src/lib/components/QueueMetricsDrawer.svelte @@ -2,6 +2,8 @@ import { Drawer, DrawerContent } from './common' import QueueMetricsDrawerInner from './QueueMetricsDrawerInner.svelte' import QueueAlerts from './QueueAlerts.svelte' + import WorkspaceFairnessEvents from './WorkspaceFairnessEvents.svelte' + import { isCloudHosted } from '$lib/cloud' let drawer: Drawer | undefined = $state() export function openDrawer() { @@ -21,6 +23,11 @@ + {#if isCloudHosted()} +
+ + {/if} +
diff --git a/frontend/src/lib/components/WorkspaceFairnessEvents.svelte b/frontend/src/lib/components/WorkspaceFairnessEvents.svelte new file mode 100644 index 0000000000..3fc1e3fc66 --- /dev/null +++ b/frontend/src/lib/components/WorkspaceFairnessEvents.svelte @@ -0,0 +1,106 @@ + + +
+ {#snippet action()} +