diff --git a/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json b/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json similarity index 83% rename from backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json rename to backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json index dabac76bfe..b0437fe128 100644 --- a/backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json +++ b/backend/.sqlx/query-a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) \n RETURNING id", + "query": "WITH uuid_table as (\n select gen_random_uuid() as uuid from generate_series(1, $11)\n )\n INSERT INTO queue \n (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s)\n (SELECT uuid, $1, $2, $3, $4, ('{ \"uuid\": \"' || uuid || '\" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13 FROM uuid_table) \n RETURNING id", "describe": { "columns": [ { @@ -63,6 +63,8 @@ "Varchar", "Timestamptz", "Varchar", + "Int4", + "Int4", "Int4" ] }, @@ -70,5 +72,5 @@ false ] }, - "hash": "6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d" + "hash": "a9915d3c72a278ff08fbe6571c2a4fc993507e255004107345b7151322e063de" } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 0fea9c4399..512f4263df 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -2283,14 +2283,22 @@ async fn add_batch_jobs( ) -> error::JsonResult> { require_super_admin(&db, &authed.email).await?; - let (hash, path, job_kind, language, dedicated_worker) = match batch_info.kind.as_str() { + let ( + hash, + path, + job_kind, + language, + dedicated_worker, + concurrent_limit, + concurrent_time_window_s, + ) = match batch_info.kind.as_str() { "script" => { if let Some(path) = batch_info.path { let ( script_hash, _tag, - _concurrent_limit, - _concurrency_time_window_s, + concurrent_limit, + concurrency_time_window_s, _cache_ttl, language, dedicated_worker, @@ -2302,6 +2310,8 @@ async fn add_batch_jobs( JobKind::Script, Some(language), dedicated_worker, + concurrent_limit, + concurrency_time_window_s, ) } else { Err(anyhow::anyhow!( @@ -2360,7 +2370,7 @@ async fn add_batch_jobs( } return Ok(Json(uuids)); } - "noop" => (None, None, JobKind::Noop, None, None), + "noop" => (None, None, JobKind::Noop, None, None, None, None), _ => { return Err(error::Error::BadRequest(format!( "Invalid batch kind: {}", @@ -2386,8 +2396,8 @@ async fn add_batch_jobs( select gen_random_uuid() as uuid from generate_series(1, $11) ) INSERT INTO queue - (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id) - (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10 FROM uuid_table) + (id, script_hash, script_path, job_kind, language, args, tag, created_by, permissioned_as, email, scheduled_for, workspace_id, concurrent_limit, concurrency_time_window_s) + (SELECT uuid, $1, $2, $3, $4, ('{ "uuid": "' || uuid || '" }')::jsonb, $5, $6, $7, $8, $9, $10, $12, $13 FROM uuid_table) RETURNING id"#, hash.map(|h| h.0), path, @@ -2399,7 +2409,9 @@ async fn add_batch_jobs( authed.email, Utc::now(), w_id, - n + n, + concurrent_limit, + concurrent_time_window_s ) .fetch_all(&db) .await?; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index d3615d850d..7af25185ac 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -1099,6 +1099,14 @@ pub async fn pull( job_custom_concurrency_time_window_s ); + sqlx::query_scalar!( + "SELECT null FROM queue WHERE id = $1 FOR UPDATE", + pulled_job.id + ) + .fetch_one(&mut tx) + .await + .context("lock job in queue")?; + let jobs_uuids_init_json_value = serde_json::from_str::( format!("{{\"{}\": {{}}}}", pulled_job.id.hyphenated().to_string()).as_str(), ) diff --git a/benchmarks/main.ts b/benchmarks/main.ts index 700be94c27..4d8d144cd4 100644 --- a/benchmarks/main.ts +++ b/benchmarks/main.ts @@ -489,7 +489,7 @@ if (import.meta.main) { } ) .option( - "--continous", + "-c --continuous", "Run the benchmark forever. This effectively disables metric collection & exports. No zombie jobs will be tracked." ) .option( diff --git a/benchmarks/pulumi/index.ts b/benchmarks/pulumi/index.ts index 24405d3bce..fa576ddb56 100644 --- a/benchmarks/pulumi/index.ts +++ b/benchmarks/pulumi/index.ts @@ -198,7 +198,7 @@ const asg = new aws.autoscaling.Group("asg3", { version: "$Latest", }, minSize: 0, - maxSize: 15, + maxSize: 30, vpcZoneIdentifiers: [subnetA.id], tags: [ { @@ -274,6 +274,8 @@ db.address.apply((address) => { environment: [ { name: "MODE", value: "worker" }, { name: "NUM_WORKERS", value: "1" }, + { name: "DATABASE_CONNECTIONS", value: "5"}, + { name: "SLEEP_QUEUE", value: "300"}, // { name: "METRICS_ADDR", value: "true" }, { name: "RUST_LOG", value: "info" }, { @@ -325,6 +327,8 @@ db.address.apply((address) => { environment: [ { name: "WORKER_GROUP", value: "dedicated" }, { name: "NUM_WORKERS", value: "10" }, + { name: "DATABASE_CONNECTIONS", value: "15"}, + { name: "SLEEP_QUEUE", value: "300"}, { name: "MODE", value: "worker" }, // { name: "METRICS_ADDR", value: "true" }, { name: "RUST_LOG", value: "info" }, @@ -372,6 +376,7 @@ db.address.apply((address) => { { name: "MODE", value: "server" }, // { name: "METRICS_ADDR", value: "true" }, { name: "RUST_LOG", value: "info" }, + { name: "DATABASE_CONNECTIONS", value: "5"}, { name: "DATABASE_URL", value: `postgres://postgres:postgres@${address}/windmill?sslmode=disable`, @@ -404,7 +409,7 @@ db.address.apply((address) => { const service_server = new aws.ecs.Service("service-server", { cluster: cluster.id, taskDefinition: server_td.arn, - desiredCount: 2, + desiredCount: 4, forceNewDeployment: true, orderedPlacementStrategies: [ { @@ -424,7 +429,7 @@ db.address.apply((address) => { const service_worker = new aws.ecs.Service("service-worker", { cluster: cluster.id, taskDefinition: worker_td.arn, - desiredCount: 10, + desiredCount: 20, forceNewDeployment: true, orderedPlacementStrategies: [ { @@ -437,7 +442,7 @@ db.address.apply((address) => { const service_worker2 = new aws.ecs.Service("service-worker-2", { cluster: cluster.id, taskDefinition: worker_td2.arn, - desiredCount: 3, + desiredCount: 30, forceNewDeployment: true, orderedPlacementStrategies: [ { diff --git a/frontend/src/lib/components/SimpleEditor.svelte b/frontend/src/lib/components/SimpleEditor.svelte index c41362f872..065176a400 100644 --- a/frontend/src/lib/components/SimpleEditor.svelte +++ b/frontend/src/lib/components/SimpleEditor.svelte @@ -76,6 +76,7 @@ code = ncode if (editor) { editor.setValue(ncode) + console.log(editor, ncode) } } diff --git a/frontend/src/lib/components/WorkspaceGroup.svelte b/frontend/src/lib/components/WorkspaceGroup.svelte index 0c1e743eaf..f9daeafd11 100644 --- a/frontend/src/lib/components/WorkspaceGroup.svelte +++ b/frontend/src/lib/components/WorkspaceGroup.svelte @@ -25,6 +25,7 @@ cache_clear?: number init_bash?: string } + export let activeWorkers: number let nconfig: any = config ? config.worker_tags != undefined || config.dedicated_worker != undefined @@ -355,7 +356,7 @@ dirty = false dirtyCode = false }} - disabled={!dirty || !$enterpriseLicense} + disabled={(!dirty && nconfig?.dedicated_worker == undefined) || !$enterpriseLicense} > Apply changes @@ -399,4 +400,9 @@ >config {JSON.stringify(config, null, 4)} {/if} + {#if activeWorkers > 1} + {activeWorkers} Workers Number of workers active in the last 10s + {/if} diff --git a/frontend/src/routes/(root)/(logged)/workers/+page.svelte b/frontend/src/routes/(root)/(logged)/workers/+page.svelte index ed121628fb..25cf19e1a9 100644 --- a/frontend/src/routes/(root)/(logged)/workers/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workers/+page.svelte @@ -20,20 +20,21 @@ let workers: WorkerPing[] | undefined = undefined let filteredWorkers: WorkerPing[] = [] let workerGroups: Record | undefined = undefined - let groupedWorkers: [string, [[string, string], WorkerPing[]][]][] = [] + let groupedWorkers: [string, [string, WorkerPing[]][]][] = [] let intervalId: NodeJS.Timer | undefined + const splitter = '_%%%_' let globalCache = false let customTags: string[] | undefined = [] $: filteredWorkers = (workers ?? []).filter((x) => (x.last_ping ?? 0) < 300) $: groupedWorkers = groupBy( groupBy( filteredWorkers, - (wp: WorkerPing) => [wp.worker_instance, wp.worker_group], + (wp: WorkerPing) => wp.worker_instance + splitter + wp.worker_group, (wp: WorkerPing) => wp.worker ), - (x) => x[0][1], - (x) => x[0][0] + (x) => x[0]?.split(splitter)?.[1], + (x) => x[0]?.split(splitter)?.[0] ) const WORKER_S3_BUCKET_SYNC_SETTING = 'worker_s3_bucket_sync' @@ -289,6 +290,9 @@ on:reload={() => { loadWorkerGroups() }} + activeWorkers={worker_group?.[1].flatMap((x) => + x[1]?.filter((y) => (y.last_ping ?? 0) < 15) + )?.length ?? 0} /> @@ -320,8 +324,11 @@ scope="colgroup" class="bg-surface-secondary/60 py-2 border-b" > - Instance: {section[0]} + Instance: {section?.split(splitter)?.[0]} IP: {workers[0].ip} + {#if workers?.length > 1} + {workers?.length} Workers + {/if}