From 07d3bb69772ae72f38a40dd13f2ae7fccab87799 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 23 Mar 2023 12:47:49 +0100 Subject: [PATCH] feat(backend): introduce RESTART_ZOMBIE_JOBS and ZOMBIE_JOB_TIMEOUT --- README.md | 73 ++++++++------- backend/sqlx-data.json | 98 ++++++++++----------- backend/src/main.rs | 4 +- backend/windmill-worker/src/worker.rs | 69 +++++++++------ frontend/src/lib/components/ArgInput.svelte | 3 + 5 files changed, 138 insertions(+), 109 deletions(-) diff --git a/README.md b/README.md index 40e0060bc0..c6cab441c0 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,8 @@ Open-source developer infrastructure for internal tools. Self-hostable alternati # Windmill - Turn scripts into workflows and UIs that you can share and run at scale -Windmill is fully open-sourced (AGPLv3) and Windmill Labs offers dedicated instance and commercial support and licenses. +Windmill is fully open-sourced (AGPLv3) and Windmill Labs offers +dedicated instance and commercial support and licenses. ![Windmill Diagram](/imgs/stacks.svg) @@ -74,7 +75,7 @@ https://user-images.githubusercontent.com/275584/218350457-bc2fdc3b-e667-4da5-a2 4. Build complex UI on top of your scripts and flows. ![Step 5](./imgs/windmill-builder.png) -Scripts and flows can also be triggered by a cron schedule '*/5 * * * *' or +Scripts and flows can also be triggered by a cron schedule '_/5 _ \* \* \*' or through webhooks. You can build your entire infra on top of Windmill! @@ -82,46 +83,49 @@ You can build your entire infra on top of Windmill! ## Show me some actual script code ```typescript -import * as wmill from "https://deno.land/x/windmill@v1.62.0/mod.ts" +import * as wmill from "https://deno.land/x/windmill@v1.62.0/mod.ts"; //import any dependency from npm -import cowsay from 'npm:cowsay@1.5.0' +import cowsay from "npm:cowsay@1.5.0"; export async function main( - a: number, - // unions generate enums - b: "my" | "enum", - // default parameters prefill the field - d = "default arg", - // nested objects work c = { nested: "object" }, - // permissioned and typed json - db: wmill.Resource<"postgresql">) { + a: number, + // unions generate enums + b: "my" | "enum", + // default parameters prefill the field + d = "default arg", + // nested objects work c = { nested: "object" }, + // permissioned and typed json + db: wmill.Resource<"postgresql"> +) { + const email = Deno.env.get("WM_EMAIL"); + // variables are permissioned and by path + let variable = await wmill.getVariable("f/company-folder/my_secret"); + const lastTimeRun = await wmill.getState(); + // logs are printed and always inspectable + console.log(cowsay.say({ text: "hello " + email + " " + lastTimeRun })); + await wmill.setState(Date.now()); - const email = Deno.env.get('WM_EMAIL') - // variables are permissioned and by path - let variable = await wmill.getVariable('f/company-folder/my_secret') - const lastTimeRun = await wmill.getState() - // logs are printed and always inspectable - console.log(cowsay.say({ text: "hello " + email + " " + lastTimeRun })) - await wmill.setState(Date.now()) - - // return is serialized as JSON - return { foo: d, variable }; + // return is serialized as JSON + return { foo: d, variable }; } ``` ## CLI We have a powerful CLI to interact with the windmill platform and sync your -scripts from local files, github repos and to run scripts and flows on the instance from local commands. See +scripts from local files, github repos and to run scripts and flows on the +instance from local commands. See [more details](https://github.com/windmill-labs/windmill/tree/main/cli) ![CLI Screencast](./cli/vhs/output/setup.gif) - ### Running scripts locally -You can run your script locally easily, you simply need to pass the right environment variables for the `wmill` client library to fetch resource and variables from your instance if necessary. See more: +You can run your script locally easily, you simply need to pass the right +environment variables for the `wmill` client library to fetch resource and +variables from your instance if necessary. See more: + ## Stack @@ -187,7 +191,6 @@ compiling from source or using without a postgres super user, see Go to http://localhost et voilà :) - The default super-admin user is: admin@windmill.dev / changeme From there, you can create other users (do not forget to change the password!) @@ -199,9 +202,9 @@ We publish helm charts at: ### Postgres without superuser -If you do not want, or cannot (for instance, in AWS Aurora or Cloud sql) use a postgres superuser, -you can run `./init-db-as-superuser.sql` to init the required users for windmill. - +If you do not want, or cannot (for instance, in AWS Aurora or Cloud sql) use a +postgres superuser, you can run `./init-db-as-superuser.sql` to init the +required users for windmill. ### Commercial license @@ -275,8 +278,8 @@ You may also add your own custom OAuth2 IdP and OAuth2 Resource provider: ### Resource types You will also want to import all the approved resource types from -[WindmillHub](https://hub.windmill.dev). A setup script will prompt -you to have it being synced automatically everyday. +[WindmillHub](https://hub.windmill.dev). A setup script will prompt you to have +it being synced automatically everyday. ## Environment Variables @@ -285,14 +288,16 @@ you to have it being synced automatically everyday. | DATABASE_URL | | The Postgres database url. | All | | DISABLE_NSJAIL | true | Disable Nsjail Sandboxing | Worker | | SERVER_BIND_ADDR | 0.0.0.0 | IP Address on which to bind listening socket | Server | -| PORT | 8000 | Exposed port | Server | | +| PORT | 8000 | Exposed port | Server | | NUM_WORKERS | 3 | The number of worker per Worker instance (set to 1 on Eks to have 1 pod = 1 worker, set to 0 for an API only instance) | Worker | | DISABLE_SERVER | false | Binary would operate as a worker only instance | Worker | | METRICS_ADDR | None | The socket addr at which to expose Prometheus metrics at the /metrics path. Set to "true" to expose it on port 8001 | All | | JSON_FMT | false | Output the logs in json format instead of logfmt | All | | BASE_URL | http://localhost:8000 | The base url that is exposed publicly to access your instance | Server | | BASE_INTERNAL_URL | http://localhost:8000 | The base url that is reachable by your workers to talk to the Servers. This help avoiding going through the external load balancer for VPC-internal requests. | Worker | -| TIMEOUT | 300 | The timeout in seconds for the execution of a script | Worker | +| TIMEOUT | 300 | The maximum time of execution of a script. When reached, the job is failed as having timedout. | Worker | +| ZOMBIE_JOB_TIMEOUT | 30 | The timeout after which a job is considered to be zombie if the worker did not send pings about processing the job (every server check for zombie jobs every 30s) | Server | +| RESTART_ZOMBIE_JOBS | true | If true then a zombie job is restarted (in-place with the same uuid and some logs), if false the zombie job is failed | Server | | SLEEP_QUEUE | 50 | The number of ms to sleep in between the last check for new jobs in the DB. It is multiplied by NUM_WORKERS such that in average, for one worker instance, there is one pull every SLEEP_QUEUE ms. | Worker | | MAX_LOG_SIZE | 500000 | The maximum number of characters a job can emit (log + result) | Worker | | DISABLE_NUSER | false | If Nsjail is enabled, disable the nsjail's `clone_newuser` setting | Worker | @@ -301,7 +306,7 @@ you to have it being synced automatically everyday. | S3_CACHE_BUCKET (EE only) | None | The S3 bucket to sync the cache of the workers to | Worker | | TAR_CACHE_RATE (EE only) | 100 | The rate at which to tar the cache of the workers. 100 means every 100th job in average (uniformly randomly distributed). | Worker | | SLACK_SIGNING_SECRET | None | The signing secret of your Slack app. See [Slack documentation](https://api.slack.com/authentication/verifying-requests-from-slack) | Server | -| COOKIE_DOMAIN | None | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin | Server | | +| COOKIE_DOMAIN | None | The domain of the cookie. If not set, the cookie will be set by the browser based on the full origin | Server | | DENO_PATH | /usr/bin/deno | The path to the deno binary. | Worker | | PYTHON_PATH | /usr/local/bin/python3 | The path to the python binary. | Worker | | GO_PATH | /usr/bin/go | The path to the go binary. | Worker | diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index ce1ccc4b57..0c082f1fbc 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -4853,6 +4853,55 @@ }, "query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *" }, + "c076993cd543ed00b4db578b9062eed512f73afc80fcd458207f17e17cde4697": { + "describe": { + "columns": [ + { + "name": "id", + "ordinal": 0, + "type_info": "Uuid" + }, + { + "name": "workspace_id", + "ordinal": 1, + "type_info": "Varchar" + }, + { + "name": "last_ping", + "ordinal": 2, + "type_info": "Timestamptz" + } + ], + "nullable": [ + false, + false, + false + ], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies" + ] + }, + "name": "job_kind" + } + } + ] + } + }, + "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping" + }, "c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc": { "describe": { "columns": [ @@ -5122,55 +5171,6 @@ }, "query": "select hash from script where path = $1 AND workspace_id = $2 AND\n created_at = (SELECT max(created_at) FROM script WHERE path = $1 AND workspace_id = $2) AND\n deleted = false" }, - "cc34f0298cfdaffe892a286e6b4390c287d33298e3342ccd1e10d48a9c3cc5c0": { - "describe": { - "columns": [ - { - "name": "id", - "ordinal": 0, - "type_info": "Uuid" - }, - { - "name": "workspace_id", - "ordinal": 1, - "type_info": "Varchar" - }, - { - "name": "last_ping", - "ordinal": 2, - "type_info": "Timestamptz" - } - ], - "nullable": [ - false, - false, - false - ], - "parameters": { - "Left": [ - "Text", - { - "Custom": { - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies" - ] - }, - "name": "job_kind" - } - } - ] - } - }, - "query": "UPDATE queue SET running = false WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping" - }, "d0308abac80575038203b60bb66d3b39b586939da0421a595e47c7a759616431": { "describe": { "columns": [], diff --git a/backend/src/main.rs b/backend/src/main.rs index f022dfecb0..eae6101e2d 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -42,7 +42,7 @@ async fn main() -> anyhow::Result<()> { let server_bind_address: IpAddr = std::env::var("SERVER_BIND_ADDR") .ok() - .and_then(|x| x.parse().ok() ) + .and_then(|x| x.parse().ok()) .unwrap_or(IpAddr::from(DEFAULT_SERVER_BIND_ADDR)); let port: u16 = std::env::var("PORT") @@ -162,6 +162,8 @@ Windmill Community Edition {GIT_VERSION} }; futures::try_join!(shutdown_signal, server_f, metrics_f, workers_f, monitor_f)?; + } else { + tracing::info!("Nothing to do, exiting."); } Ok(()) } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b4bf236ab4..afa3d9f716 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -331,9 +331,8 @@ lazy_static::lazy_static! { .ok() .map(|x| format!(";{x}")) .unwrap_or_else(|| String::new()); - static ref NPM_CONFIG_REGISTRY: Option = std::env::var("NPM_CONFIG_REGISTRY").ok(); - + static ref NPM_CONFIG_REGISTRY: Option = std::env::var("NPM_CONFIG_REGISTRY").ok(); static ref DENO_FLAGS: Option> = std::env::var("DENO_FLAGS") .ok() @@ -391,10 +390,19 @@ lazy_static::lazy_static! { .ok() .and_then(|x| x.parse::().ok()) .unwrap_or(DEFAULT_TIMEOUT as u16); - + static ref TIMEOUT_DURATION: Duration = Duration::from_secs(*TIMEOUT as u64); - static ref ZOMBIE_JOB_TIMEOUT: String = (*TIMEOUT as u32 * 5).to_string(); + static ref ZOMBIE_JOB_TIMEOUT: String = std::env::var("ZOMBIE_JOB_TIMEOUT") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or_else(|| "30".to_string()); + + + pub static ref RESTART_ZOMBIE_JOBS: bool = std::env::var("RESTART_ZOMBIE_JOBS") + .ok() + .and_then(|x| x.parse::().ok()) + .unwrap_or(true); static ref SESSION_TOKEN_EXPIRY: i32 = (*TIMEOUT as i32) * 2; } @@ -881,6 +889,9 @@ async fn handle_queued_job( } _ => { let mut logs = "".to_string(); + if let Some(log_str) = &job.logs { + logs.push_str(&log_str); + } if job.is_flow_step { update_flow_status_in_progress( @@ -2710,7 +2721,7 @@ pub async fn handle_zombie_jobs_periodically( handle_zombie_jobs(db, base_internal_url).await; tokio::select! { - _ = tokio::time::sleep(Duration::from_secs(60)) => (), + _ = tokio::time::sleep(Duration::from_secs(2)) => (), _ = rx.recv() => { println!("received killpill for monitor job"); break; @@ -2720,28 +2731,34 @@ pub async fn handle_zombie_jobs_periodically( } async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { - let restarted = sqlx::query!( - "UPDATE queue SET running = false WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping", - *ZOMBIE_JOB_TIMEOUT, - JobKind::Flow: JobKind, - ) - .fetch_all(db) - .await - .ok() - .unwrap_or_else(|| vec![]); + if *RESTART_ZOMBIE_JOBS { + let restarted = sqlx::query!( + "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping", + *ZOMBIE_JOB_TIMEOUT, + JobKind::Flow: JobKind, + ) + .fetch_all(db) + .await + .ok() + .unwrap_or_else(|| vec![]); - QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); - for r in restarted { - tracing::info!( - "restarted zombie job {} {} {}", - r.id, - r.workspace_id, - r.last_ping - ); + QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _); + for r in restarted { + tracing::info!( + "restarted zombie job {} {} {}", + r.id, + r.workspace_id, + r.last_ping + ); + } } + let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2".to_string(); + if *RESTART_ZOMBIE_JOBS { + timeout_query.push_str(" same_worker = true"); + }; let timeouts = sqlx::query_as::<_, QueuedJob>( - "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = true", + &timeout_query ) .bind(ZOMBIE_JOB_TIMEOUT.as_str()) .bind(JobKind::Flow) @@ -2753,7 +2770,7 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _); for job in timeouts { tracing::info!( - "timedouts zombie same_worker job {} {}", + "timedout zombie job {} {}", job.id, job.workspace_id, ); @@ -2775,11 +2792,13 @@ async fn handle_zombie_jobs(db: &Pool, base_internal_url: &str) { tx.commit().await.expect("could not commit job token"); let client = AuthedClient { base_internal_url: base_internal_url.to_string(), token: token.clone(), workspace: job.workspace_id.to_string(), client: OnceCell::new() }; + let last_ping = job.last_ping.clone(); let _ = handle_job_error( db, &client, job, - error::Error::ExecutionErr("Same worker job timed out".to_string()), + error::Error::ExecutionErr(format!("Job timed out after no ping from job since {} (ZOMBIE_JOB_TIMEOUT: {})", + last_ping.map(|x| x.to_string()).unwrap_or_else(|| "no ping".to_string()), *ZOMBIE_JOB_TIMEOUT)), None, true, same_worker_tx_never_used, diff --git a/frontend/src/lib/components/ArgInput.svelte b/frontend/src/lib/components/ArgInput.svelte index 84dbc8d229..ee5999bf8b 100644 --- a/frontend/src/lib/components/ArgInput.svelte +++ b/frontend/src/lib/components/ArgInput.svelte @@ -188,6 +188,7 @@ use:autosize rows="1" bind:value={description} + on:keydown|stopPropagation placeholder="Field description" /> {#if type == 'string' && format != 'date-time'} @@ -339,6 +340,7 @@ {autofocus} {disabled} use:autosize + on:keydown|stopPropagation style="max-height: {maxHeight}" on:input={() => { dispatch('input', { rawValue: value, isRaw: false }) @@ -414,6 +416,7 @@ }} on:blur={() => dispatch('blur')} use:autosize + on:keydown|stopPropagation type="text" {disabled} class="col-span-10 {valid