audit-log workspace-fairness cap transitions (#9306)

* feat(queue): audit-log workspace-fairness cap transitions

When the cloud per-workspace fairness mechanism adds a workspace to the
capped set or releases one, write `workspace_fairness.capped` /
`workspace_fairness.uncapped` audit-log entries to the affected workspace.
The cluster admin can review the full timeline from the `admins` workspace
audit view with `all_workspaces=true`; per-workspace owners see their own
events in their normal audit list.

Only the per-cycle refresh winner emits entries (matching where the heavy
aggregation runs), so a fleet of N workers does not produce N duplicates
per transition. The diff is computed against the value already in
`background_task_state` rather than the winner's in-memory cache, so a
freshly-restarted process winning the claim does not spuriously emit
"newly capped" entries for workspaces that were already capped before it
started.

Audit writes are best-effort: failures are logged via tracing and do not
abort the refresh cycle.

Fixes WIN-1984

* feat(queue): scope fairness audit to admins workspace + queue-metrics pane

- Write `workspace_fairness.capped` / `workspace_fairness.uncapped` to the
  `admins` workspace (was: per-affected-workspace) with the affected
  workspace_id moved to the `resource` field. Cluster admins now get the
  full timeline in one place without `all_workspaces=true`.
- Add `GET /workers/workspace_fairness_events` returning the last 100
  events. Cloud-gated (returns `[]` on non-cloud) and devops-only.
- Add a `WorkspaceFairnessEvents` Section to the Queue Metrics drawer,
  rendered only when `isCloudHosted()` is true. Shows time / event
  badge / workspace / parameters with a refresh button.

Fixes WIN-1984
This commit is contained in:
Ruben Fiszel
2026-06-08 11:35:45 +02:00
committed by tristantr
parent b5a17bdcb7
commit a91af58324
6 changed files with 368 additions and 1 deletions
@@ -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<Postgres>) -> 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<String> = if won {
let prev_val: Option<serde_json::Value> =
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::<FairnessState>(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<Postgres>) -> 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<Postgres>) -> 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<Postgres>,
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:#}");
}
}
}