af02e200a5
Add new BENCHMARK_KIND variants (sequentialflow, scriptlogs, concurrencylimit, concurrencykey, mixed, mixed_no_cc) for targeted performance testing. Fix shared iteration counting across workers using a global atomic counter. Add job_perms inserts and queue diagnostics for benchmark mode. Move db connection setup to dedicated module and drop the initial connection pool before creating the main one, preventing connection starvation when PostgreSQL max_connections is low. Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
912 lines
44 KiB
Rust
912 lines
44 KiB
Rust
use crate::{
|
|
worker::{write_file, TMP_DIR},
|
|
DB,
|
|
};
|
|
use serde::Serialize;
|
|
use std::collections::HashSet;
|
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
use tokio::time::Instant;
|
|
use uuid::Uuid;
|
|
|
|
static BENCHMARK_INITIALIZED: AtomicBool = AtomicBool::new(false);
|
|
|
|
static SHARED_BENCH_ITERS: std::sync::LazyLock<Arc<AtomicU64>> =
|
|
std::sync::LazyLock::new(|| Arc::new(AtomicU64::new(0)));
|
|
|
|
pub fn shared_bench_iters() -> Arc<AtomicU64> {
|
|
SHARED_BENCH_ITERS.clone()
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct PoolStats {
|
|
pub peak_active_conns: u32,
|
|
pub pool_saturation_histogram: Vec<u32>,
|
|
}
|
|
|
|
impl PoolStats {
|
|
pub fn new(pool_size: u32) -> Self {
|
|
PoolStats {
|
|
peak_active_conns: 0,
|
|
pool_saturation_histogram: vec![0; pool_size as usize + 1],
|
|
}
|
|
}
|
|
|
|
pub fn sample(&mut self, pool_size: u32, num_idle: u32) {
|
|
let active = pool_size.saturating_sub(num_idle);
|
|
if active > self.peak_active_conns {
|
|
self.peak_active_conns = active;
|
|
}
|
|
if let Some(bucket) = self.pool_saturation_histogram.get_mut(active as usize) {
|
|
*bucket += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct BenchmarkInfo {
|
|
#[serde(skip)]
|
|
pub start: Instant,
|
|
#[serde(skip)]
|
|
pub iters: u64,
|
|
#[serde(skip)]
|
|
seen_top_level: HashSet<Uuid>,
|
|
#[serde(skip)]
|
|
pub shared_iters: Arc<AtomicU64>,
|
|
timings: Vec<BenchmarkIter>,
|
|
pub iter_durations: Vec<u64>,
|
|
pub total_duration: Option<u64>,
|
|
pub pool_stats: Option<PoolStats>,
|
|
}
|
|
|
|
impl BenchmarkInfo {
|
|
pub fn new(shared_iters: Arc<AtomicU64>) -> Self {
|
|
BenchmarkInfo {
|
|
iters: 0,
|
|
seen_top_level: HashSet::new(),
|
|
shared_iters,
|
|
timings: vec![],
|
|
start: Instant::now(),
|
|
iter_durations: vec![],
|
|
total_duration: None,
|
|
pool_stats: None,
|
|
}
|
|
}
|
|
|
|
pub fn init_pool_stats(&mut self, pool_size: u32) {
|
|
self.pool_stats = Some(PoolStats::new(pool_size));
|
|
}
|
|
|
|
pub fn sample_pool(&mut self, pool_size: u32, num_idle: u32) {
|
|
if let Some(stats) = self.pool_stats.as_mut() {
|
|
stats.sample(pool_size, num_idle);
|
|
}
|
|
}
|
|
|
|
pub fn count_top_level(&mut self, job_id: Uuid) -> bool {
|
|
if self.seen_top_level.insert(job_id) {
|
|
self.iters += 1;
|
|
return true;
|
|
}
|
|
false
|
|
}
|
|
|
|
pub fn add_iter(&mut self, bench: BenchmarkIter, job_id: Uuid, is_top_level: bool) -> bool {
|
|
let newly_counted = is_top_level && self.seen_top_level.insert(job_id);
|
|
if newly_counted {
|
|
self.iters += 1;
|
|
}
|
|
let elapsed_total = bench.start.elapsed().as_nanos() as u64;
|
|
self.timings.push(bench);
|
|
self.iter_durations.push(elapsed_total);
|
|
newly_counted
|
|
}
|
|
|
|
pub fn write_to_file(&mut self, path: &str) -> anyhow::Result<()> {
|
|
let total_duration = self.start.elapsed().as_millis() as u64;
|
|
self.total_duration = Some(total_duration as u64);
|
|
|
|
let pool_info = self.pool_stats.as_ref().map_or(String::new(), |ps| {
|
|
format!(", peak active conns: {}", ps.peak_active_conns)
|
|
});
|
|
println!(
|
|
"Writing benchmark {path}, duration of benchmark: {total_duration}ms and RPS: {}{pool_info}",
|
|
self.iters as f64 / total_duration as f64 * 1000.0
|
|
);
|
|
write_file(TMP_DIR, path, &serde_json::to_string(&self).unwrap()).expect("write profiling");
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct BenchmarkIter {
|
|
#[serde(skip)]
|
|
pub start: Instant,
|
|
#[serde(skip)]
|
|
last_instant: Instant,
|
|
last_step: String,
|
|
timings: Vec<(String, u32)>,
|
|
}
|
|
|
|
impl BenchmarkIter {
|
|
pub fn new() -> Self {
|
|
BenchmarkIter {
|
|
last_instant: Instant::now(),
|
|
timings: vec![],
|
|
start: Instant::now(),
|
|
last_step: String::new(),
|
|
}
|
|
}
|
|
|
|
pub fn add_timing(&mut self, name: &str) {
|
|
let elapsed = self.last_instant.elapsed().as_nanos() as u32;
|
|
self.timings
|
|
.push((format!("{}->{}", self.last_step, name), elapsed));
|
|
self.last_instant = Instant::now();
|
|
self.last_step = name.to_string();
|
|
}
|
|
}
|
|
|
|
pub async fn benchmark_verify(benchmark_jobs: i32, db: &DB) {
|
|
let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string());
|
|
|
|
if benchmark_jobs <= 0 || benchmark_kind == "none" {
|
|
return;
|
|
}
|
|
|
|
// For flows, child jobs are created dynamically so only check top-level (parent_job IS NULL).
|
|
// "parallelflow" inserts only 1 top-level flow regardless of benchmark_jobs.
|
|
let expected_top_level = match benchmark_kind.as_str() {
|
|
"parallelflow" => 1i64,
|
|
_ => benchmark_jobs as i64,
|
|
};
|
|
|
|
let row = sqlx::query!(
|
|
"SELECT
|
|
COUNT(*) FILTER (WHERE status = 'success') AS succeeded,
|
|
COUNT(*) FILTER (WHERE status = 'failure') AS failed,
|
|
COUNT(*) FILTER (WHERE status = 'canceled') AS canceled
|
|
FROM v2_job_completed
|
|
JOIN v2_job USING (id)
|
|
WHERE v2_job.workspace_id = 'admins' AND v2_job.parent_job IS NULL",
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.expect("benchmark verify query failed");
|
|
|
|
let succeeded = row.succeeded.unwrap_or(0);
|
|
let failed = row.failed.unwrap_or(0);
|
|
let canceled = row.canceled.unwrap_or(0);
|
|
let total = succeeded + failed + canceled;
|
|
|
|
let remaining_in_queue = sqlx::query_scalar!(
|
|
"SELECT COUNT(*) FROM v2_job_queue WHERE workspace_id = 'admins'",
|
|
)
|
|
.fetch_one(db)
|
|
.await
|
|
.expect("benchmark verify queue query failed")
|
|
.unwrap_or(0);
|
|
|
|
println!("=== BENCHMARK VERIFICATION ===");
|
|
println!(" kind: {benchmark_kind}");
|
|
println!(" expected top-level: {expected_top_level}");
|
|
println!(" completed total: {total}");
|
|
println!(" succeeded: {succeeded}");
|
|
println!(" failed: {failed}");
|
|
println!(" canceled: {canceled}");
|
|
println!(" still in queue: {remaining_in_queue}");
|
|
|
|
if failed > 0 || canceled > 0 {
|
|
tracing::error!(
|
|
"BENCHMARK VERIFICATION FAILED: {failed} failed, {canceled} canceled out of {total} completed"
|
|
);
|
|
}
|
|
if remaining_in_queue > 0 {
|
|
tracing::warn!(
|
|
"BENCHMARK VERIFICATION: {remaining_in_queue} jobs still in queue after benchmark"
|
|
);
|
|
}
|
|
if succeeded != expected_top_level {
|
|
tracing::error!(
|
|
"BENCHMARK VERIFICATION FAILED: expected {expected_top_level} succeeded top-level jobs, got {succeeded}"
|
|
);
|
|
} else if failed == 0 && canceled == 0 && remaining_in_queue == 0 {
|
|
println!(" result: ALL PASSED");
|
|
}
|
|
println!("==============================");
|
|
}
|
|
|
|
pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
|
|
use crate::{jobs::JobKind, scripts::ScriptLang};
|
|
|
|
// Only the first worker to reach this point runs init
|
|
if BENCHMARK_INITIALIZED.swap(true, Ordering::SeqCst) {
|
|
return;
|
|
}
|
|
|
|
let benchmark_kind = std::env::var("BENCHMARK_KIND").unwrap_or("noop".to_string());
|
|
|
|
if benchmark_jobs > 0 {
|
|
// Clean up data from previous benchmark runs
|
|
sqlx::query!("DELETE FROM v2_job_completed WHERE workspace_id = 'admins'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up v2_job_completed: {e:#}"));
|
|
sqlx::query!("DELETE FROM v2_job_queue WHERE workspace_id = 'admins'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up v2_job_queue: {e:#}"));
|
|
sqlx::query!("DELETE FROM v2_job_status WHERE id IN (SELECT id FROM v2_job WHERE workspace_id = 'admins')")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up v2_job_status: {e:#}"));
|
|
sqlx::query!("DELETE FROM v2_job_runtime WHERE id IN (SELECT id FROM v2_job WHERE workspace_id = 'admins')")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up v2_job_runtime: {e:#}"));
|
|
sqlx::query("DELETE FROM job_perms WHERE workspace_id = 'admins'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up job_perms: {e:#}"));
|
|
sqlx::query!("DELETE FROM concurrency_key WHERE key LIKE 'bench_%' OR key LIKE 'u/admin/bench_%'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up concurrency_key: {e:#}"));
|
|
sqlx::query!("DELETE FROM concurrency_counter WHERE concurrency_id LIKE 'bench_%' OR concurrency_id LIKE 'u/admin/bench_%'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up concurrency_counter: {e:#}"));
|
|
sqlx::query!("DELETE FROM v2_job WHERE workspace_id = 'admins'")
|
|
.execute(db)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to clean up v2_job: {e:#}"));
|
|
|
|
let mut tx = db.begin().await.unwrap();
|
|
match benchmark_kind.as_str() {
|
|
"dedicated" => {
|
|
// you need to create the script first, check https://github.com/windmill-labs/windmill/blob/b76a92cfe454c686f005c65f534e29e039f3c706/benchmarks/lib.ts#L47
|
|
let hash = sqlx::query_scalar!(
|
|
"SELECT hash FROM script WHERE path = $1 AND workspace_id = $2",
|
|
"f/benchmarks/dedicated",
|
|
"admins"
|
|
)
|
|
.fetch_one(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs"));
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
|
hash,
|
|
"f/benchmarks/dedicated",
|
|
JobKind::Script as JobKind,
|
|
ScriptLang::Bun as ScriptLang,
|
|
"admins:f/benchmarks/dedicated",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
benchmark_jobs
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "admins:f/benchmarks/dedicated")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert dedicated jobs (3)"));
|
|
}
|
|
"parallelflow" => {
|
|
//create dedicated script
|
|
sqlx::query!("INSERT INTO script (summary, description, dedicated_worker, content, workspace_id, path, hash, language, tag, created_by, lock) VALUES ('', '', true, $1, $2, $3, $4, $5, $6, $7, '') ON CONFLICT (workspace_id, hash) DO NOTHING",
|
|
"export async function main() {
|
|
console.log('hello world');
|
|
}",
|
|
"admins",
|
|
"u/admin/parallelflow",
|
|
1234567890,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"flow",
|
|
"admin",
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs {_e:#}"));
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, 1)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::FlowPreview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"flow",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
serde_json::from_str::<serde_json::Value>(r#"
|
|
{
|
|
"modules": [
|
|
{
|
|
"id": "a",
|
|
"value": {
|
|
"type": "forloopflow",
|
|
"modules": [
|
|
{
|
|
"id": "b",
|
|
"value": {
|
|
"path": "u/admin/parallelflow",
|
|
"type": "script",
|
|
"tag_override": "",
|
|
"input_transforms": {}
|
|
},
|
|
"summary": "calctest"
|
|
}
|
|
],
|
|
"iterator": {
|
|
"expr": "[...new Array(300)]",
|
|
"type": "javascript"
|
|
},
|
|
"parallel": true,
|
|
"parallelism": 10,
|
|
"skip_failures": true
|
|
}
|
|
}
|
|
],
|
|
"preprocessor_module": null
|
|
}
|
|
"#).unwrap(),
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "flow")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (3)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
|
|
&uuids,
|
|
serde_json::from_str::<serde_json::Value>(
|
|
r#"
|
|
{
|
|
"step": 0,
|
|
"modules": [
|
|
{
|
|
"id": "a",
|
|
"type": "WaitingForPriorSteps"
|
|
}
|
|
],
|
|
"cleanup_module": {},
|
|
"failure_module": {
|
|
"id": "failure",
|
|
"type": "WaitingForPriorSteps"
|
|
},
|
|
"preprocessor_module": null
|
|
}
|
|
|
|
"#
|
|
)
|
|
.unwrap()
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert parallelflow jobs (4)"));
|
|
}
|
|
"sequentialflow" => {
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::FlowPreview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"flow",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
serde_json::from_str::<serde_json::Value>(r#"
|
|
{
|
|
"modules": [
|
|
{
|
|
"id": "a",
|
|
"value": {
|
|
"type": "rawscript",
|
|
"content": "export async function main() { return 'a'; }",
|
|
"language": "deno",
|
|
"input_transforms": {}
|
|
}
|
|
},
|
|
{
|
|
"id": "b",
|
|
"value": {
|
|
"type": "rawscript",
|
|
"content": "export async function main() { return 'b'; }",
|
|
"language": "deno",
|
|
"input_transforms": {}
|
|
}
|
|
},
|
|
{
|
|
"id": "c",
|
|
"value": {
|
|
"type": "rawscript",
|
|
"content": "export async function main() { return 'c'; }",
|
|
"language": "deno",
|
|
"input_transforms": {}
|
|
}
|
|
},
|
|
{
|
|
"id": "d",
|
|
"value": {
|
|
"type": "rawscript",
|
|
"content": "export async function main() { return 'd'; }",
|
|
"language": "deno",
|
|
"input_transforms": {}
|
|
}
|
|
}
|
|
],
|
|
"preprocessor_module": null
|
|
}
|
|
"#).unwrap(),
|
|
benchmark_jobs,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert sequentialflow jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "flow")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert sequentialflow jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert sequentialflow jobs (3)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
|
|
&uuids,
|
|
serde_json::from_str::<serde_json::Value>(
|
|
r#"
|
|
{
|
|
"step": 0,
|
|
"modules": [
|
|
{ "id": "a", "type": "WaitingForPriorSteps" },
|
|
{ "id": "b", "type": "WaitingForPriorSteps" },
|
|
{ "id": "c", "type": "WaitingForPriorSteps" },
|
|
{ "id": "d", "type": "WaitingForPriorSteps" }
|
|
],
|
|
"cleanup_module": {},
|
|
"failure_module": {
|
|
"id": "failure",
|
|
"type": "WaitingForPriorSteps"
|
|
},
|
|
"preprocessor_module": null
|
|
}
|
|
"#
|
|
)
|
|
.unwrap()
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert sequentialflow jobs (4)"));
|
|
}
|
|
"scriptlogs" => {
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { for (let i = 0; i < 1000; i++) { console.log('benchmark log line ' + i); } return 'done'; }",
|
|
benchmark_jobs,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert scriptlogs jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert scriptlogs jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert scriptlogs jobs (3)"));
|
|
}
|
|
"concurrencylimit" => {
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code, concurrent_limit, concurrency_time_window_s) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, $13)) RETURNING id",
|
|
None::<i64>,
|
|
Some("u/admin/bench_conclimit"),
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { return 'done'; }",
|
|
2i32,
|
|
0i32,
|
|
benchmark_jobs,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert concurrencylimit jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert concurrencylimit jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencylimit jobs (3)"));
|
|
let concurrency_id = "u/admin/bench_conclimit";
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
|
|
concurrency_id,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencylimit counter"));
|
|
for uuid in &uuids {
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_key (key, ended_at, job_id) VALUES ($1, now() - INTERVAL '1 hour', $2)",
|
|
concurrency_id,
|
|
uuid,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencylimit key"));
|
|
}
|
|
}
|
|
"concurrencykey" => {
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code, concurrent_limit, concurrency_time_window_s) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, $13)) RETURNING id",
|
|
None::<i64>,
|
|
Some("u/admin/bench_conckey"),
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { return 'done'; }",
|
|
1i32,
|
|
0i32,
|
|
benchmark_jobs,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert concurrencykey jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert concurrencykey jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencykey jobs (3)"));
|
|
let concurrency_id = "bench_shared_concurrency_key";
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
|
|
concurrency_id,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencykey counter"));
|
|
for uuid in &uuids {
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_key (key, ended_at, job_id) VALUES ($1, now() - INTERVAL '1 hour', $2)",
|
|
concurrency_id,
|
|
uuid,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert concurrencykey key"));
|
|
}
|
|
}
|
|
"mixed" => {
|
|
let portion = benchmark_jobs / 5;
|
|
let remainder = benchmark_jobs % 5;
|
|
|
|
// 1) noop jobs
|
|
let noop_count = portion + remainder;
|
|
let noop_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Noop as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
noop_count,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed noop jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &noop_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed noop queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &noop_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed noop runtime"));
|
|
|
|
// 2) sequentialflow jobs
|
|
if portion > 0 {
|
|
let sf_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::FlowPreview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"flow",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
serde_json::from_str::<serde_json::Value>(r#"{"modules":[{"id":"a","value":{"type":"rawscript","content":"export async function main() { return 'a'; }","language":"deno","input_transforms":{}}},{"id":"b","value":{"type":"rawscript","content":"export async function main() { return 'b'; }","language":"deno","input_transforms":{}}},{"id":"c","value":{"type":"rawscript","content":"export async function main() { return 'c'; }","language":"deno","input_transforms":{}}},{"id":"d","value":{"type":"rawscript","content":"export async function main() { return 'd'; }","language":"deno","input_transforms":{}}}],"preprocessor_module":null}"#).unwrap(),
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sf_uuids, "admins", "flow")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sf_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow runtime"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
|
|
&sf_uuids,
|
|
serde_json::from_str::<serde_json::Value>(r#"{"step":0,"modules":[{"id":"a","type":"WaitingForPriorSteps"},{"id":"b","type":"WaitingForPriorSteps"},{"id":"c","type":"WaitingForPriorSteps"},{"id":"d","type":"WaitingForPriorSteps"}],"cleanup_module":{},"failure_module":{"id":"failure","type":"WaitingForPriorSteps"},"preprocessor_module":null}"#).unwrap()
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow status"));
|
|
}
|
|
|
|
// 3) scriptlogs jobs
|
|
if portion > 0 {
|
|
let sl_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { for (let i = 0; i < 1000; i++) { console.log('benchmark log line ' + i); } return 'done'; }",
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sl_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sl_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs runtime"));
|
|
}
|
|
|
|
// 4) concurrencylimit jobs
|
|
if portion > 0 {
|
|
let cl_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code, concurrent_limit, concurrency_time_window_s) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, $13)) RETURNING id",
|
|
None::<i64>,
|
|
Some("u/admin/bench_conclimit"),
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { return 'done'; }",
|
|
2i32,
|
|
0i32,
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &cl_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &cl_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit runtime"));
|
|
let cl_concurrency_id = "u/admin/bench_conclimit";
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
|
|
cl_concurrency_id,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit counter"));
|
|
for uuid in &cl_uuids {
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_key (key, ended_at, job_id) VALUES ($1, now() - INTERVAL '1 hour', $2)",
|
|
cl_concurrency_id,
|
|
uuid,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit key"));
|
|
}
|
|
}
|
|
|
|
// 5) concurrencykey jobs
|
|
if portion > 0 {
|
|
let ck_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code, concurrent_limit, concurrency_time_window_s) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12 FROM generate_series(1, $13)) RETURNING id",
|
|
None::<i64>,
|
|
Some("u/admin/bench_conckey"),
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { return 'done'; }",
|
|
1i32,
|
|
0i32,
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &ck_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &ck_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey runtime"));
|
|
let ck_concurrency_id = "bench_shared_concurrency_key";
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
|
|
ck_concurrency_id,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey counter"));
|
|
for uuid in &ck_uuids {
|
|
sqlx::query!(
|
|
"INSERT INTO concurrency_key (key, ended_at, job_id) VALUES ($1, now() - INTERVAL '1 hour', $2)",
|
|
ck_concurrency_id,
|
|
uuid,
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey key"));
|
|
}
|
|
}
|
|
}
|
|
"mixed_no_cc" => {
|
|
let portion = benchmark_jobs / 3;
|
|
let remainder = benchmark_jobs % 3;
|
|
|
|
// 1) noop jobs
|
|
let noop_count = portion + remainder;
|
|
let noop_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Noop as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
noop_count,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &noop_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &noop_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop runtime"));
|
|
|
|
// 2) sequentialflow jobs
|
|
if portion > 0 {
|
|
let sf_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_flow) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::FlowPreview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"flow",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
serde_json::from_str::<serde_json::Value>(r#"{"modules":[{"id":"a","value":{"type":"rawscript","content":"export async function main() { return 'a'; }","language":"deno","input_transforms":{}}},{"id":"b","value":{"type":"rawscript","content":"export async function main() { return 'b'; }","language":"deno","input_transforms":{}}},{"id":"c","value":{"type":"rawscript","content":"export async function main() { return 'c'; }","language":"deno","input_transforms":{}}},{"id":"d","value":{"type":"rawscript","content":"export async function main() { return 'd'; }","language":"deno","input_transforms":{}}}],"preprocessor_module":null}"#).unwrap(),
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sf_uuids, "admins", "flow")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sf_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow runtime"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
|
|
&sf_uuids,
|
|
serde_json::from_str::<serde_json::Value>(r#"{"step":0,"modules":[{"id":"a","type":"WaitingForPriorSteps"},{"id":"b","type":"WaitingForPriorSteps"},{"id":"c","type":"WaitingForPriorSteps"},{"id":"d","type":"WaitingForPriorSteps"}],"cleanup_module":{},"failure_module":{"id":"failure","type":"WaitingForPriorSteps"},"preprocessor_module":null}"#).unwrap()
|
|
)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow status"));
|
|
}
|
|
|
|
// 3) scriptlogs jobs
|
|
if portion > 0 {
|
|
let sl_uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id, raw_code) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 FROM generate_series(1, $11)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Preview as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
"export async function main() { for (let i = 0; i < 1000; i++) { console.log('benchmark log line ' + i); } return 'done'; }",
|
|
portion,
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs jobs"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sl_uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs queue"));
|
|
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sl_uuids)
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs runtime"));
|
|
}
|
|
}
|
|
"none" => {}
|
|
_ => {
|
|
let uuids = sqlx::query_scalar!("INSERT INTO v2_job (id, runnable_id, runnable_path, kind, script_lang, tag, created_by, permissioned_as, permissioned_as_email, workspace_id) (SELECT gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7, $8, $9 FROM generate_series(1, $10)) RETURNING id",
|
|
None::<i64>,
|
|
None::<String>,
|
|
JobKind::Noop as JobKind,
|
|
ScriptLang::Deno as ScriptLang,
|
|
"deno",
|
|
"admin",
|
|
"u/admin",
|
|
"admin@windmill.dev",
|
|
"admins",
|
|
benchmark_jobs
|
|
)
|
|
.fetch_all(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (1)"));
|
|
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &uuids, "admins", "deno")
|
|
.execute(&mut *tx)
|
|
.await.unwrap_or_else(|_e| panic!("failed to insert noop jobs (2)"));
|
|
sqlx::query!(
|
|
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
|
|
&uuids
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|_e| panic!("failed to insert noop jobs (3)"));
|
|
}
|
|
}
|
|
// Insert job_perms for all benchmark jobs so workers don't fall back to slow permission lookups
|
|
sqlx::query(
|
|
"INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, groups, folders, workspace_id)
|
|
SELECT id, 'admin@windmill.dev', 'admin', true, false, ARRAY['all']::text[], ARRAY[]::jsonb[], 'admins'
|
|
FROM v2_job WHERE workspace_id = 'admins'"
|
|
)
|
|
.execute(&mut *tx)
|
|
.await
|
|
.unwrap_or_else(|e| panic!("failed to insert job_perms: {e:#}"));
|
|
|
|
tx.commit().await.unwrap();
|
|
}
|
|
}
|