diff --git a/backend/migrations/20230705102646_add_error_handler_workspace.down.sql b/backend/migrations/20230705102646_add_error_handler_workspace.down.sql new file mode 100644 index 0000000000..56d6c562cf --- /dev/null +++ b/backend/migrations/20230705102646_add_error_handler_workspace.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +ALTER TABLE workspace_settings DROP COLUMN error_handler; \ No newline at end of file diff --git a/backend/migrations/20230705102646_add_error_handler_workspace.up.sql b/backend/migrations/20230705102646_add_error_handler_workspace.up.sql new file mode 100644 index 0000000000..f5fe2d7b06 --- /dev/null +++ b/backend/migrations/20230705102646_add_error_handler_workspace.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TABLE workspace_settings ADD COLUMN error_handler VARCHAR(1000); \ No newline at end of file diff --git a/backend/sqlx-data.json b/backend/sqlx-data.json index 6aa85670ab..1239af4798 100644 --- a/backend/sqlx-data.json +++ b/backend/sqlx-data.json @@ -702,6 +702,11 @@ "name": "deploy_to", "ordinal": 10, "type_info": "Varchar" + }, + { + "name": "error_handler", + "ordinal": 11, + "type_info": "Varchar" } ], "nullable": [ @@ -715,6 +720,7 @@ true, true, true, + true, true ], "parameters": { @@ -1222,7 +1228,9 @@ "deno", "go", "bash", - "postgresql" + "postgresql", + "nativets", + "bun" ] }, "name": "script_lang" @@ -2106,7 +2114,9 @@ "deno", "go", "bash", - "postgresql" + "postgresql", + "nativets", + "bun" ] }, "name": "script_lang" @@ -2512,6 +2522,11 @@ "name": "deploy_to", "ordinal": 10, "type_info": "Varchar" + }, + { + "name": "error_handler", + "ordinal": 11, + "type_info": "Varchar" } ], "nullable": [ @@ -2525,6 +2540,7 @@ true, true, true, + true, true ], "parameters": { @@ -2774,7 +2790,9 @@ "deno", "go", "bash", - "postgresql" + "postgresql", + "nativets", + "bun" ] }, "name": "script_lang" @@ -3882,7 +3900,9 @@ "deno", "go", "bash", - "postgresql" + "postgresql", + "nativets", + "bun" ] }, "name": "script_lang" @@ -5624,6 +5644,18 @@ }, "query": "UPDATE schedule SET script_path = $1 WHERE script_path = $2 AND workspace_id = $3 AND is_flow IS false RETURNING *" }, + "c0635f65d561c1a6b183b8fea45320482375d0bb5e617a9dd236a631a7b9ff05": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Text" + ] + } + }, + "query": "UPDATE workspace_settings SET error_handler = NULL WHERE workspace_id = $1" + }, "c07c9276945663d062cf0ff5b3323be681a0e2cb07a457ea9aede2daeff551cc": { "describe": { "columns": [ @@ -6673,6 +6705,19 @@ }, "query": "DELETE FROM password WHERE email = $1" }, + "e844b9ee75a1b5438ffe743185fc78f649db89c78b2b8cd7e778bf18a6b41e67": { + "describe": { + "columns": [], + "nullable": [], + "parameters": { + "Left": [ + "Varchar", + "Text" + ] + } + }, + "query": "UPDATE workspace_settings SET error_handler = $1 WHERE workspace_id = $2" + }, "e9c0e331c16312bf086b17c91466c5389d41454fd3f18d73c2e9554845ee9a72": { "describe": { "columns": [], @@ -7223,6 +7268,26 @@ }, "query": "SELECT (flow_status->'step')::integer as step, jsonb_array_length(flow_status->'modules') as len FROM queue WHERE id = $1" }, + "f8b34e09453d51d3df5be20652938a890ee353fec64ee73c312a3352da7f7515": { + "describe": { + "columns": [ + { + "name": "error_handler", + "ordinal": 0, + "type_info": "Varchar" + } + ], + "nullable": [ + true + ], + "parameters": { + "Left": [ + "Text" + ] + } + }, + "query": "SELECT error_handler FROM workspace_settings WHERE workspace_id = $1" + }, "fa258894bd90ea6586669e5810c5c6bcb42d5e1e68fab27fb185d06962b6454a": { "describe": { "columns": [ diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c674a47262..82db201a06 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -971,6 +971,8 @@ paths: type: string deploy_to: type: string + error_handler: + type: string /w/{workspace}/workspaces/get_deploy_to: get: @@ -1123,6 +1125,33 @@ paths: schema: type: string + /w/{workspace}/workspaces/edit_error_handler: + post: + summary: edit error handler + operationId: editErrorHandler + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: WorkspaceErrorHandler + required: true + content: + application/json: + schema: + type: object + properties: + error_handler: + type: string + + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + /w/{workspace}/users/list: get: summary: list users diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 353b00748c..36282b8f32 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -34,6 +34,7 @@ use magic_crypt::MagicCryptTrait; #[cfg(feature = "enterprise")] use stripe::CustomerId; use windmill_audit::{audit_log, ActionKind}; +use windmill_common::users::username_to_permissioned_as; use windmill_common::{ error::{to_anyhow, Error, JsonResult, Result}, flows::Flow, @@ -64,7 +65,8 @@ pub fn workspaced_service() -> Router { .route("/edit_auto_invite", post(edit_auto_invite)) .route("/edit_deploy_to", post(edit_deploy_to)) .route("/tarball", get(tarball_workspace)) - .route("/premium_info", get(premium_info)); + .route("/premium_info", get(premium_info)) + .route("/edit_error_handler", post(edit_error_handler)); #[cfg(feature = "enterprise")] tracing::info!("stripe enabled"); @@ -111,6 +113,7 @@ pub struct WorkspaceSettings { pub plan: Option, pub webhook: Option, pub deploy_to: Option, + pub error_handler: Option, } #[derive(FromRow, Serialize, Debug)] @@ -202,6 +205,11 @@ pub struct NewWorkspaceUser { pub operator: bool, } +#[derive(Deserialize)] +pub struct EditErrorHandler { + pub error_handler: Option, +} + async fn list_pending_invites( authed: Authed, Extension(user_db): Extension, @@ -644,6 +652,62 @@ async fn edit_webhook( Ok(format!("Edit webhook for workspace {}", &w_id)) } + +async fn edit_error_handler( + authed: Authed, + Extension(db): Extension, + Path(w_id): Path, + Authed { is_admin, username, .. }: Authed, + Json(ee): Json, +) -> Result { + require_admin(is_admin, &username)?; + + let mut tx = db.begin().await?; + + sqlx::query_as!( + Group, + "INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES ($1, $2, $3, $4) ON CONFLICT DO NOTHING", + w_id, + "error_handler", + "The group the error handler acts on belhalf of", + serde_json::json!({username_to_permissioned_as(&authed.username): true}) + ) + .execute(&mut tx) + .await?; + + if let Some(error_handler) = &ee.error_handler { + + sqlx::query!( + "UPDATE workspace_settings SET error_handler = $1 WHERE workspace_id = $2", + error_handler, + &w_id + ) + .execute(&mut tx) + .await?; + } else { + sqlx::query!( + "UPDATE workspace_settings SET error_handler = NULL WHERE workspace_id = $1", + &w_id, + ) + .execute(&mut tx) + .await?; + } + audit_log( + &mut tx, + &authed.username, + "workspaces.edit_error_handler", + ActionKind::Update, + &w_id, + Some(&authed.email), + Some([("error_handler", &format!("{:?}", ee.error_handler)[..])].into()), + ) + .await?; + tx.commit().await?; + + Ok(format!("Edit error_handler for workspace {}", &w_id)) + +} + async fn list_workspaces_as_super_admin( authed: Authed, Extension(user_db): Extension, diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 738c748b83..f0e4dc3d83 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -8,6 +8,7 @@ use std::{collections::HashMap, vec}; +use anyhow::Context; use async_recursion::async_recursion; use itertools::Itertools; use reqwest::Client; @@ -369,19 +370,88 @@ pub async fn add_completed_job( && queued_job.parent_job.is_none() && !success { - if let Err(e) = send_error_to_global_handler(rsmq, &queued_job, db, &result).await { + if let Err(e) = send_error_to_global_handler(rsmq.clone(), &queued_job, db, &result).await { tracing::error!( "Could not run global error handler for job {}: {}", &queued_job.id, e ); } + + if let Err(e) = + send_error_to_workspace_handler(rsmq.clone(), &queued_job, db, &result).await + { + tracing::error!( + "Could not run workspace error handler for job {}: {}", + &queued_job.id, + e + ); + } } tracing::debug!("Added completed job {}", queued_job.id); Ok(queued_job.id) } +pub async fn run_error_handler( + rsmq: Option, + queued_job: &QueuedJob, + db: &Pool, + result: &serde_json::Value, + error_handler_path: &str, + is_global: bool, +) -> Result<(), Error> { + let w_id = &queued_job.workspace_id; + let script_w_id = if is_global { "admins" } else { w_id }; // script workspace id + let job_id = queued_job.id; + let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into(); + let (job_payload, tag) = + script_path_to_payload(&error_handler_path, tx.transaction_mut(), script_w_id).await?; + let mut args = result.as_object().unwrap().clone(); + args.insert("workspace_id".to_string(), json!(w_id)); + args.insert("job_id".to_string(), json!(job_id)); + args.insert("path".to_string(), json!(queued_job.script_path)); + args.insert("is_flow".to_string(), json!(queued_job.raw_flow.is_some())); + args.insert("email".to_string(), json!(queued_job.email)); + + let (uuid, tx) = push( + tx, + script_w_id, + job_payload, + args, + if is_global { "global" } else { "error_handler" }, + if is_global { + SUPERADMIN_SECRET_EMAIL + } else { + "error_handler@windmill.dev" + }, + if is_global { + SUPERADMIN_SECRET_EMAIL.to_string() + } else { + "g/error_handler".to_string() + }, + None, + None, + Some(job_id), + Some(job_id), + None, + false, + false, + None, + true, + tag, + ) + .await?; + tx.commit().await?; + + let error_handler_type = if is_global { "global" } else { "workspace" }; + tracing::info!( + "Sent error of job {job_id} to {error_handler_type} error handler under uuid {uuid}" + ); + + Ok(()) +} + pub async fn send_error_to_global_handler( rsmq: Option, queued_job: &QueuedJob, @@ -389,43 +459,44 @@ pub async fn send_error_to_global_handler Result<(), Error> { if let Some(ref global_error_handler) = *GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE { - let w_id = &queued_job.workspace_id; - let job_id = queued_job.id; - let mut tx: QueueTransaction<'_, _> = (rsmq, db.begin().await?).into(); - let (job_payload, tag) = - script_path_to_payload(&global_error_handler, tx.transaction_mut(), "admins").await?; - let mut args = result.as_object().unwrap().clone(); - args.insert("workspace_id".to_string(), json!(w_id)); - args.insert("job_id".to_string(), json!(job_id)); - args.insert("path".to_string(), json!(queued_job.script_path)); - args.insert("is_flow".to_string(), json!(queued_job.raw_flow.is_some())); - args.insert("email".to_string(), json!(queued_job.email)); - - let (uuid, tx) = push( - tx, - "admins", - job_payload, - args, - "global", - SUPERADMIN_SECRET_EMAIL, - SUPERADMIN_SECRET_EMAIL.to_string(), - None, - None, - Some(job_id), - Some(job_id), - None, - false, - false, - None, - true, - tag, - ) - .await?; - tx.commit().await?; - tracing::info!("Sent error of job {job_id} to global error handler under uuid {uuid}"); + run_error_handler(rsmq, queued_job, db, result, global_error_handler, true).await? } + Ok(()) } + +pub async fn send_error_to_workspace_handler( + rsmq: Option, + queued_job: &QueuedJob, + db: &Pool, + result: &serde_json::Value, +) -> Result<(), Error> { + let w_id = &queued_job.workspace_id; + let mut tx = db.begin().await?; + let error_handler = sqlx::query_scalar!( + "SELECT error_handler FROM workspace_settings WHERE workspace_id = $1", + &w_id + ) + .fetch_optional(&mut tx) + .await + .context("sending error to global handler")? + .ok_or_else(|| Error::InternalErr(format!("no workspace settings for id {w_id}")))?; + + if let Some(error_handler) = error_handler { + run_error_handler( + rsmq, + queued_job, + db, + result, + &error_handler.strip_prefix("script/").unwrap(), + false, + ) + .await? + } + + Ok(()) +} + #[instrument(level = "trace", skip_all)] pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clone + Send + 'c>( mut tx: QueueTransaction<'c, R>, diff --git a/frontend/src/lib/components/ScheduleEditor.svelte b/frontend/src/lib/components/ScheduleEditor.svelte index 7a15bbfa40..87be9fe68e 100644 --- a/frontend/src/lib/components/ScheduleEditor.svelte +++ b/frontend/src/lib/components/ScheduleEditor.svelte @@ -249,12 +249,7 @@ {#if runnable} {#if runnable?.schema && runnable.schema.properties && Object.keys(runnable.schema.properties).length > 0} - + {:else}
This {is_flow ? 'flow' : 'script'} takes no argument @@ -267,6 +262,7 @@ {/if}

Error Handler

+ -
The following args will be passed to the error handler: -
    -
  • path: The path of the script or flow that errored
    • -
  • schedule_path: The path of the schedule
    • -
  • error: The error details
    • -
+
+
The following args will be passed to the error handler: +
    +
  • path: The path of the script or flow that errored
    • +
  • schedule_path: The path of the schedule
    • +
  • error: The error details
    • +
+
+
diff --git a/frontend/src/lib/components/ScriptPicker.svelte b/frontend/src/lib/components/ScriptPicker.svelte index 061131c6f9..67db1c1902 100644 --- a/frontend/src/lib/components/ScriptPicker.svelte +++ b/frontend/src/lib/components/ScriptPicker.svelte @@ -15,6 +15,8 @@ import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte' import { Code2, Globe } from 'lucide-svelte' import type { SupportedLanguage } from '$lib/common' + import { faRotateRight } from '@fortawesome/free-solid-svg-icons' + import Icon from 'svelte-awesome' import FlowIcon from './home/FlowIcon.svelte' export let initialPath: string | undefined = undefined @@ -24,6 +26,7 @@ export let itemKind: 'hub' | 'script' | 'flow' = allowHub ? 'hub' : 'script' export let kind: Script.kind = Script.kind.SCRIPT export let disabled = false + export let canRefresh = false let items: { value: string; label: string }[] = [] let drawerViewer: Drawer @@ -107,6 +110,12 @@ /> {/if} + {#if canRefresh} + + {/if} + {#if scriptPath !== undefined && scriptPath !== ''} {#if itemKind == 'flow'} + {:else if tab == 'error_handler'} + + +
+
+
The following args will be passed to the error handler: +
    +
  • path: The path of the script or flow that errored
    • +
  • email: The email of the user who ran the script or flow that errored
    • +
  • error: The error details
    • +
  • job_id: The job id
    • +
  • is_flow: Whether the error comes from a flow
    • +
  • workspace_id: The workspace id of the failed script or flow
    • +
+
+ The error handler will be executed by the automatically created group g/error_handler. If + your error handler requires variables or resources, you need to add them to the group. +
+
+
+
+
{/if} {:else}