feat: workspace error handler (#1799)

* feat: add workspace error handler * feat: run error handler as group * fix: handler picker initial path * fix(backend): separate global / workspace handlers * fix(frontend): error handler picker tab change --------- Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
2023-07-06 21:43:15 +02:00
parent c677c6d79a
commit 69ee98a835
9 changed files with 361 additions and 54 deletions
--- a/backend/migrations/20230705102646_add_error_handler_workspace.down.sql
+++ 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;
--- a/backend/migrations/20230705102646_add_error_handler_workspace.up.sql
+++ 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);
--- 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": [
--- 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
--- 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<String>,
    pub webhook: Option<String>,
    pub deploy_to: Option<String>,
+    pub error_handler: Option<String>,
 }

 #[derive(FromRow, Serialize, Debug)]
@@ -202,6 +205,11 @@ pub struct NewWorkspaceUser {
    pub operator: bool,
 }

+#[derive(Deserialize)]
+pub struct EditErrorHandler {
+    pub error_handler: Option<String>,
+}
+
 async fn list_pending_invites(
    authed: Authed,
    Extension(user_db): Extension<UserDB>,
@@ -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<DB>,
+    Path(w_id): Path<String>,
+    Authed { is_admin, username, .. }: Authed,
+    Json(ee): Json<EditErrorHandler>,
+) -> Result<String> {
+    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<UserDB>,
--- 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<R: rsmq_async::RsmqConnection + Clone + Send>(
        && 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<R: rsmq_async::RsmqConnection + Clone + Send>(
+    rsmq: Option<R>,
+    queued_job: &QueuedJob,
+    db: &Pool<Postgres>,
+    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<R: rsmq_async::RsmqConnection + Clone + Send>(
    rsmq: Option<R>,
    queued_job: &QueuedJob,
@@ -389,43 +459,44 @@ pub async fn send_error_to_global_handler<R: rsmq_async::RsmqConnection + Clone
    result: &serde_json::Value,
 ) -> 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<R: rsmq_async::RsmqConnection + Clone + Send>(
+    rsmq: Option<R>,
+    queued_job: &QueuedJob,
+    db: &Pool<Postgres>,
+    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>,
--- 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}
-						<SchemaForm
-							disabled={!can_write}
-							schema={runnable.schema}
-							bind:isValid
-							bind:args
-						/>
+						<SchemaForm disabled={!can_write} schema={runnable.schema} bind:isValid bind:args />
 					{:else}
 						<div class="text-xs texg-gray-700">
 							This {is_flow ? 'flow' : 'script'} takes no argument
@@ -267,6 +262,7 @@
 				{/if}
 			</div>
 			<h2 class="border-b pb-1 mt-8 mb-2">Error Handler</h2>
+
 			<ScriptPicker
 				disabled={initialScriptPath != '' || !can_write}
 				initialPath={errorHandlerPath}
@@ -274,14 +270,22 @@
 				allowFlow={true}
 				bind:scriptPath={errorHandlerPath}
 				bind:itemKind={errorHandleritemKind}
+				canRefresh
 			/>
-			<div class="text-gray-600 italic text-sm mt-2"
-				>The following args will be passed to the error handler:
-				<ul class="mt-1 ml-2">
-					<li><b>path</b>: The path of the script or flow that errored</li>
-					<li><b>schedule_path</b>: The path of the schedule</li>
-					<li><b>error</b>: The error details</li>
-				</ul>
+			<div class="flex gap-20 items-start mt-3">
+				<div class="text-gray-600 italic text-sm"
+					>The following args will be passed to the error handler:
+					<ul class="mt-1 ml-2">
+						<li><b>path</b>: The path of the script or flow that errored</li>
+						<li><b>schedule_path</b>: The path of the schedule</li>
+						<li><b>error</b>: The error details</li>
+					</ul>
+				</div>
+				<Button
+					wrapperClasses="mt-6"
+					href="/scripts/add?hub=hub%2F1087%2Fwindmill%2Fschedule_error_handler_template"
+					target="_blank">Use template</Button
+				>
 			</div>
 		</div>
 	</DrawerContent>
--- 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}
+		<Button variant="border" color="light" wrapperClasses="self-stretch" on:click={loadItems}
+			><Icon scale={0.8} data={faRotateRight} /></Button
+		>
+	{/if}
+
 	{#if scriptPath !== undefined && scriptPath !== ''}
 		{#if itemKind == 'flow'}
 			<Button
--- a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
+++ b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
@@ -53,6 +53,9 @@
 	let customer_id: string | undefined = undefined
 	let webhook: string | undefined = undefined
 	let workspaceToDeployTo: string | undefined = undefined
+	let errorHandlerInitialPath: string
+	let errorHandlerScriptPath: string
+	let errorHandlerItemKind: 'script' = 'script'
 	let tab =
 		($page.url.searchParams.get('tab') as
 			| 'users'
@@ -60,7 +63,8 @@
 			| 'premium'
 			| 'export_delete'
 			| 'webhook'
-			| 'deploy_to') ?? 'users'
+			| 'deploy_to'
+			| 'error_handler') ?? 'users'

 	// function getDropDownItems(username: string): DropdownItem[] {
 	// 	return [
@@ -135,6 +139,8 @@
 		customer_id = settings.customer_id
 		workspaceToDeployTo = settings.deploy_to
 		webhook = settings.webhook
+		errorHandlerScriptPath = (settings.error_handler ?? '').split('/').slice(1).join('/')
+		errorHandlerInitialPath = errorHandlerScriptPath
 	}

 	async function listUsers(): Promise<void> {
@@ -185,6 +191,23 @@
 		)
 	}

+	async function editErrorHandler() {
+		errorHandlerInitialPath = errorHandlerScriptPath
+		if (errorHandlerScriptPath) {
+			await WorkspaceService.editErrorHandler({
+				workspace: $workspaceStore!,
+				requestBody: { error_handler: `${errorHandlerItemKind}/${errorHandlerScriptPath}` }
+			})
+			sendUserToast(`workspace error handler set to ${errorHandlerScriptPath}`)
+		} else {
+			await WorkspaceService.editErrorHandler({
+				workspace: $workspaceStore!,
+				requestBody: { error_handler: undefined }
+			})
+			sendUserToast(`workspace error handler removed`)
+		}
+	}
+
 	const plans = {
 		Free: [
 			'Users use their individual global free-tier quotas when doing executions in this workspace',
@@ -249,6 +272,9 @@
 						<div class="flex gap-2 items-center my-1">Webhook for CLI Sync</div>
 					</Tab>
 				{/if}
+				<Tab size="md" value="error_handler">
+					<div class="flex gap-2 items-center my-1">Error Handler</div>
+				</Tab>
 			</Tabs>
 		</div>
 		{#if tab == 'users'}
@@ -787,6 +813,41 @@
 					>Set Webhook</Button
 				>
 			</div>
+		{:else if tab == 'error_handler'}
+			<PageHeader title="Script to run as error handler" primary={false} />
+			<ScriptPicker
+				kind={Script.kind.SCRIPT}
+				bind:itemKind={errorHandlerItemKind}
+				bind:scriptPath={errorHandlerScriptPath}
+				initialPath={errorHandlerInitialPath}
+				on:select={editErrorHandler}
+				canRefresh
+			/>
+			<div class="flex gap-20 items-start mt-3">
+				<div class="w-2/3">
+					<div class="text-gray-600 italic text-sm"
+						>The following args will be passed to the error handler:
+						<ul class="mt-1 ml-2">
+							<li><b>path</b>: The path of the script or flow that errored</li>
+							<li><b>email</b>: The email of the user who ran the script or flow that errored</li>
+							<li><b>error</b>: The error details</li>
+							<li><b>job_id</b>: The job id</li>
+							<li><b>is_flow</b>: Whether the error comes from a flow</li>
+							<li><b>workspace_id</b>: The workspace id of the failed script or flow</li>
+						</ul>
+						<br />
+						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.
+					</div>
+				</div>
+				<div class="w-1/3 flex items-start">
+					<Button
+						wrapperClasses="mt-6"
+						href="/scripts/add?hub=hub%2F1088%2Fwindmill%2FGlobal_%2F_workspace_error_handler_template"
+						target="_blank">Use template</Button
+					></div
+				>
+			</div>
 		{/if}
 	{:else}
 		<div class="bg-red-100 border-l-4 border-red-600 text-orange-700 p-4 m-4" role="alert">