starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Schedule error handler improvements (#2555)

Browse Source 
* Fix schedule path for scheduled scripts

* feat: Add possibility to mute workspace error handler on schedules and define an error/recovery handler as default

* Permission error handlers to the error handler group for consistency, and small refactoring

* Fix recovery handler

* Add parent job to recovery handler for consistency

* Fix tests

* Locking feature to EE in FE
This commit is contained in:
Guillaume Bouvignies
2023-11-04 00:52:37 +01:00
committed by GitHub
parent e3de82058b
commit d3698fc07b
22 changed files with 341 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-0be97088e8ca1e8be9bf4ea45f830cceedd598a3221c522a1f9026d2d48a0ffb.json → backend/.sqlx/query-2157813137b383411227b06814143df2d2187d1d43b84f67d22c0a2365b42ada.json
Generated
+10 -3
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10 WHERE path = $11 AND workspace_id = $12 RETURNING *",
"query": "UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10, ws_error_handler_muted = $11 \n WHERE path = $12 AND workspace_id = $13 RETURNING *",
"describe": {
"columns": [
{
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -116,6 +121,7 @@
"Varchar",
"Int4",
"Json",
"Bool",
"Text",
"Text"
]
@@ -140,8 +146,9 @@
true,
true,
true,
true
true,
false
]
},
"hash": "0be97088e8ca1e8be9bf4ea45f830cceedd598a3221c522a1f9026d2d48a0ffb"
"hash": "2157813137b383411227b06814143df2d2187d1d43b84f67d22c0a2365b42ada"
}
backend/.sqlx/query-9b9afb9d5ef57b70de1207c4546b5547e579d978d899325d821a5aee1df1a9fc.json → backend/.sqlx/query-47a01c0924df7bc2695f8238425aa7eadf845e4156da8cf645582737e046589d.json
Generated
+11 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING *",
"query": "INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, ws_error_handler_muted) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) RETURNING *",
"describe": {
"columns": [
{
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -122,7 +127,8 @@
"Json",
"Varchar",
"Int4",
"Json"
"Json",
"Bool"
]
},
"nullable": [
@@ -145,8 +151,9 @@
true,
true,
true,
true
true,
false
]
},
"hash": "9b9afb9d5ef57b70de1207c4546b5547e579d978d899325d821a5aee1df1a9fc"
"hash": "47a01c0924df7bc2695f8238425aa7eadf845e4156da8cf645582737e046589d"
}
backend/.sqlx/query-7a00843921db4ac1eb20a1045e83363587c1a2cb62d2df25d5564ae78a03b097.json
Generated
+6
View File
@@ -105,6 +105,11 @@
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
},
{
"ordinal": 21,
"name": "jobs",
"type_info": "JsonArray"
}
@@ -137,6 +142,7 @@
true,
true,
true,
false,
null
]
},
backend/.sqlx/query-911b1e1f2a5ba6d5159916e5598020e680c45043b0736ad0153ee261a151dd90.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -131,7 +136,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "911b1e1f2a5ba6d5159916e5598020e680c45043b0736ad0153ee261a151dd90"
backend/.sqlx/query-96dc1cd308f3c21cfb50b88048054dc03f93e261d25969d66aa48e9d0502960f.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -132,7 +137,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "96dc1cd308f3c21cfb50b88048054dc03f93e261d25969d66aa48e9d0502960f"
backend/.sqlx/query-aa2800113a8a8805f47cdc1dd0f29d94c546fe531e7edd3e91da4978af5442fb.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -130,7 +135,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "aa2800113a8a8805f47cdc1dd0f29d94c546fe531e7edd3e91da4978af5442fb"
backend/.sqlx/query-c10348d26e3e3dc9e345d9044db35db9906617eb9c98aaf58a55d9681ffb85e3.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -131,7 +136,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "c10348d26e3e3dc9e345d9044db35db9906617eb9c98aaf58a55d9681ffb85e3"
backend/.sqlx/query-dd74fa9468b5fe8c8ad657ded06076c11a78d0206af2af20685c691d4d9520bb.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -131,7 +136,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "dd74fa9468b5fe8c8ad657ded06076c11a78d0206af2af20685c691d4d9520bb"
backend/.sqlx/query-eced0a09ba547ce1dccb54a5419b22373603c9d01f77047b3553bde125bf71e8.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -131,7 +136,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "eced0a09ba547ce1dccb54a5419b22373603c9d01f77047b3553bde125bf71e8"
backend/.sqlx/query-fdd3710a381dac33ef5ee5ec5564a4874e6d7807d43fa6ea7b8408cad9e97480.json
Generated
+7 -1
View File
@@ -102,6 +102,11 @@
"ordinal": 19,
"name": "on_recovery_extra_args",
"type_info": "Json"
},
{
"ordinal": 20,
"name": "ws_error_handler_muted",
"type_info": "Bool"
}
],
"parameters": {
@@ -129,7 +134,8 @@
true,
true,
true,
true
true,
false
]
},
"hash": "fdd3710a381dac33ef5ee5ec5564a4874e6d7807d43fa6ea7b8408cad9e97480"
backend/migrations/20231103143535_schedule_default_error_handler.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
ALTER TABLE schedule DROP COLUMN ws_error_handler_muted;
backend/migrations/20231103143535_schedule_default_error_handler.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE schedule ADD COLUMN ws_error_handler_muted BOOLEAN NOT NULL DEFAULT false;
backend/tests/worker.rs
+4
View File
@@ -2854,6 +2854,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
script_path: "f/system/failing_script".to_string(),
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
};
let _ = client.create_schedule("test-workspace", &schedule).await;
@@ -2913,6 +2914,7 @@ async fn test_script_schedule_handlers(db: Pool<Postgres>) {
on_recovery_extra_args: None,
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
},
)
.await
@@ -2988,6 +2990,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
script_path: "f/system/failing_flow".to_string(),
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
};
let _ = client.create_schedule("test-workspace", &schedule).await;
@@ -3048,6 +3051,7 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) {
on_recovery_extra_args: None,
timezone: "UTC".to_string(),
schedule: format!("{} {} * * * *", then.second(), then.minute()).to_string(),
ws_error_handler_muted: None,
},
)
.await
backend/windmill-api/openapi-deref.yaml
+12 -2
View File
@@ -1,6 +1,6 @@
openapi: 3.0.3
info:
version: 1.192.0
version: 1.197.1
title: Windmill API
contact:
name: Windmill Team
@@ -3048,13 +3048,17 @@ paths:
type: array
items:
type: string
/embeddings/query_resource_types:
/w/{workspace}/embeddings/query_resource_types:
get:
summary: query resource types by similarity
operationId: queryResourceTypes
tags:
- resource
parameters:
- name: workspace
in: path
required: true
schema: *ref_0
- name: text
description: query text
in: query
@@ -7818,6 +7822,8 @@ paths:
on_recovery_extra_args:
type: object
additionalProperties: *ref_14
ws_error_handler_muted:
type: boolean
required: &ref_133
- path
- schedule
@@ -7878,6 +7884,8 @@ paths:
on_recovery_extra_args:
type: object
additionalProperties: *ref_14
ws_error_handler_muted:
type: boolean
required: &ref_135
- schedule
- timezone
@@ -8014,6 +8022,8 @@ paths:
on_recovery_extra_args:
type: object
additionalProperties: *ref_14
ws_error_handler_muted:
type: boolean
required: &ref_74
- path
- edited_by
backend/windmill-api/openapi.yaml
+6
View File
@@ -7580,6 +7580,8 @@ components:
type: number
on_recovery_extra_args:
$ref: "#/components/schemas/ScriptArgs"
ws_error_handler_muted:
type: boolean
required:
- path
- edited_by
@@ -7645,6 +7647,8 @@ components:
type: number
on_recovery_extra_args:
$ref: "#/components/schemas/ScriptArgs"
ws_error_handler_muted:
type: boolean
required:
- path
- schedule
@@ -7677,6 +7681,8 @@ components:
type: number
on_recovery_extra_args:
$ref: "#/components/schemas/ScriptArgs"
ws_error_handler_muted:
type: boolean
required:
- schedule
- timezone
backend/windmill-api/src/schedule.rs
+10 -4
View File
@@ -62,6 +62,7 @@ pub struct NewSchedule {
pub on_recovery: Option<String>,
pub on_recovery_times: Option<i32>,
pub on_recovery_extra_args: Option<serde_json::Value>,
pub ws_error_handler_muted: Option<bool>,
}
async fn check_path_conflict<'c>(
@@ -105,7 +106,8 @@ async fn create_schedule(
#[cfg(not(feature = "enterprise"))]
if ns.on_failure.is_some()
&& ns.on_failure.as_ref().unwrap() == "script/hub/2431/slack/schedule-error-handler-slack"
&& ns.on_failure.as_ref().unwrap()
== "script/hub/5792/workspace-or-schedule-error-handler-slack"
{
return Err(Error::BadRequest(
"Slack error handler is only available in enterprise version".to_string(),
@@ -136,7 +138,7 @@ async fn create_schedule(
let schedule = sqlx::query_as!(
Schedule,
"INSERT INTO schedule (workspace_id, path, schedule, timezone, edited_by, script_path, \
is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) RETURNING *",
is_flow, args, enabled, email, on_failure, on_failure_times, on_failure_exact, on_failure_extra_args, on_recovery, on_recovery_times, on_recovery_extra_args, ws_error_handler_muted) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) RETURNING *",
w_id,
ns.path,
ns.schedule,
@@ -154,6 +156,7 @@ async fn create_schedule(
ns.on_recovery,
ns.on_recovery_times,
ns.on_recovery_extra_args,
ns.ws_error_handler_muted.unwrap_or(false),
)
.fetch_one(&mut tx)
.await
@@ -215,8 +218,8 @@ async fn edit_schedule(
clear_schedule(tx.transaction_mut(), path, is_flow, &w_id).await?;
let schedule = sqlx::query_as!(
Schedule,
"UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10 WHERE path \
= $11 AND workspace_id = $12 RETURNING *",
"UPDATE schedule SET schedule = $1, timezone = $2, args = $3, on_failure = $4, on_failure_times = $5, on_failure_exact = $6, on_failure_extra_args = $7, on_recovery = $8, on_recovery_times = $9, on_recovery_extra_args = $10, ws_error_handler_muted = $11
WHERE path = $12 AND workspace_id = $13 RETURNING *",
es.schedule,
es.timezone,
es.args,
@@ -227,6 +230,7 @@ async fn edit_schedule(
es.on_recovery,
es.on_recovery_times,
es.on_recovery_extra_args,
es.ws_error_handler_muted.unwrap_or(false),
path,
w_id,
)
@@ -317,6 +321,7 @@ pub struct ScheduleWJobs {
pub on_recovery: Option<String>,
pub on_recovery_times: Option<i32>,
pub on_recovery_extra_args: Option<serde_json::Value>,
pub ws_error_handler_muted: bool,
pub jobs: Option<Vec<serde_json::Value>>,
}
@@ -519,6 +524,7 @@ pub struct EditSchedule {
pub on_recovery: Option<String>,
pub on_recovery_times: Option<i32>,
pub on_recovery_extra_args: Option<serde_json::Value>,
pub ws_error_handler_muted: Option<bool>,
}
pub async fn clear_schedule<'c>(
backend/windmill-api/src/workspaces.rs
+3 -3
View File
@@ -35,6 +35,7 @@ use chrono::Utc;
use magic_crypt::MagicCryptTrait;
#[cfg(feature = "enterprise")]
use stripe::CustomerId;
use uuid::Uuid;
use windmill_audit::{audit_log, ActionKind};
use windmill_common::db::UserDB;
use windmill_common::schedule::Schedule;
@@ -543,7 +544,8 @@ async fn run_slack_message_test_job(
let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into();
let (uuid, tx) = windmill_queue::handle_on_failure(
&db,
tx,
tx,
Uuid::parse_str("00000000-0000-0000-0000-000000000000")?,
"slack_message_test",
"slack_message_test",
false,
@@ -553,9 +555,7 @@ async fn run_slack_message_test_job(
0,
Utc::now(),
Some(json!(extra_args)),
authed.username.as_str(),
authed.email.as_str(),
username_to_permissioned_as(authed.username.as_str()),
None, // Note: we could mark it as high priority to return result quickly to the user
).await?;
tx.commit().await?;
backend/windmill-common/src/schedule.rs
+1
View File
@@ -33,6 +33,7 @@ pub struct Schedule {
pub on_recovery: Option<String>,
pub on_recovery_times: Option<i32>,
pub on_recovery_extra_args: Option<serde_json::Value>,
pub ws_error_handler_muted: bool,
}
pub fn schedule_to_user(path: &str) -> String {
backend/windmill-queue/src/jobs.rs
+73 -35
View File
@@ -54,9 +54,9 @@ use windmill_common::{
JobPayload, QueuedJob, RawCode,
},
oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH,
schedule::{schedule_to_user, Schedule},
schedule::Schedule,
scripts::{ScriptHash, ScriptLang},
users::{username_to_permissioned_as, SUPERADMIN_SECRET_EMAIL},
users::SUPERADMIN_SECRET_EMAIL,
worker::{to_raw_value, WORKER_CONFIG},
DB, METRICS_ENABLED,
};
@@ -103,6 +103,10 @@ const MAX_FREE_EXECS: i32 = 1000;
#[cfg(feature = "enterprise")]
const MAX_FREE_CONCURRENT_RUNS: i32 = 15;
const ERROR_HANDLER_USERNAME: &str = "error_handler";
const ERROR_HANDLER_USER_GROUP: &str = "g/error_handler";
const ERROR_HANDLER_USER_EMAIL: &str = "error_handler@windmill.dev";
#[async_recursion]
pub async fn cancel_job<'c: 'async_recursion>(
username: &str,
@@ -422,6 +426,7 @@ pub async fn add_completed_job<
// tracing::error!("2 {:?}", start.elapsed());
// tracing::error!("Added completed job {:#?}", queued_job);
let mut skip_downstream_error_handlers = false;
tx = delete_job(tx, &queued_job.workspace_id, job_id).await?;
// tracing::error!("3 {:?}", start.elapsed());
@@ -429,7 +434,7 @@ pub async fn add_completed_job<
&& queued_job.schedule_path.is_some()
&& queued_job.script_path.is_some()
{
tx = apply_schedule_handlers(
(skip_downstream_error_handlers, tx) = apply_schedule_handlers(
tx,
db,
queued_job.schedule_path.as_ref().unwrap(),
@@ -496,7 +501,8 @@ pub async fn add_completed_job<
.map_err(|e| Error::InternalErr(format!("updating usage: {e}")));
}
if matches!(queued_job.job_kind, JobKind::Flow | JobKind::Script)
if !skip_downstream_error_handlers
&& matches!(queued_job.job_kind, JobKind::Flow | JobKind::Script)
&& queued_job.parent_job.is_none()
&& !success
{
@@ -550,8 +556,16 @@ pub async fn run_error_handler<
"is_flow".to_string(),
to_raw_value(&queued_job.raw_flow.is_some()),
);
extra.insert(
"started_at".to_string(),
to_raw_value(&queued_job.started_at),
);
extra.insert("email".to_string(), to_raw_value(&queued_job.email));
if let Some(schedule_path) = &queued_job.schedule_path {
extra.insert("schedule_path".to_string(), to_raw_value(schedule_path));
}
if let Some(extra_args) = error_handler_extra_args {
if let serde_json::Value::Object(args_m) = extra_args {
for (k, v) in args_m {
@@ -563,11 +577,13 @@ pub async fn run_error_handler<
));
}
}
// TODO: this should be injected when the EH is defined in the BE.
if error_handler_path
.to_string()
.eq("hub/2431/slack/schedule-error-handler-slack")
.eq("hub/5792/workspace-or-schedule-error-handler-slack")
{
// custom slack error handler being used -> we need to inject the slack token
// default slack error handler being used -> we need to inject the slack token
let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}");
extra.insert("slack".to_string(), to_raw_value(&slack_resource));
}
@@ -580,16 +596,20 @@ pub async fn run_error_handler<
script_w_id,
job_payload,
PushArgs { extra, args: result.to_owned() },
if is_global { "global" } else { "error_handler" },
if is_global {
"global"
} else {
ERROR_HANDLER_USERNAME
},
if is_global {
SUPERADMIN_SECRET_EMAIL
} else {
"error_handler@windmill.dev"
ERROR_HANDLER_USER_EMAIL
},
if is_global {
SUPERADMIN_SECRET_EMAIL.to_string()
} else {
"g/error_handler".to_string()
ERROR_HANDLER_USER_GROUP.to_string()
},
None,
None,
@@ -749,6 +769,7 @@ pub async fn handle_maybe_scheduled_job<'c, R: rsmq_async::RsmqConnection + Clon
on_recovery: schedule.on_recovery,
on_recovery_times: schedule.on_recovery_times,
on_recovery_extra_args: schedule.on_recovery_extra_args,
ws_error_handler_muted: schedule.ws_error_handler_muted,
},
)
.await;
@@ -794,17 +815,18 @@ async fn apply_schedule_handlers<
job_id: Uuid,
started_at: DateTime<Utc>,
job_priority: Option<i16>,
) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
) -> windmill_common::error::Result<(bool, QueueTransaction<'c, R>)> {
let schedule = get_schedule_opt(tx.transaction_mut(), w_id, schedule_path).await?;
if schedule.is_none() {
tracing::error!(
"Schedule {schedule_path} in {w_id} not found. Impossible to apply schedule handlers"
);
return Ok(tx);
return Ok((false, tx));
}
let schedule = schedule.unwrap();
let skip_downstream_error_handlers = schedule.ws_error_handler_muted;
if !success {
if let Some(on_failure_path) = schedule.on_failure.clone() {
@@ -831,13 +853,14 @@ async fn apply_schedule_handlers<
};
if !match_times {
return Ok(tx);
return Ok((skip_downstream_error_handlers, tx));
}
}
let on_failure_result = handle_on_failure(
db,
tx,
job_id,
schedule_path,
script_path,
schedule.is_flow,
@@ -848,8 +871,6 @@ async fn apply_schedule_handlers<
started_at,
schedule.on_failure_extra_args,
&schedule.email,
&schedule_to_user(&schedule.path),
username_to_permissioned_as(&schedule.edited_by),
job_priority,
)
.await;
@@ -890,13 +911,13 @@ async fn apply_schedule_handlers<
).fetch_all(&mut tx).await?;
if past_jobs.len() < times as usize {
return Ok(tx);
return Ok((skip_downstream_error_handlers, tx));
}
let n_times_successful = past_jobs[..(times - 1) as usize].iter().all(|j| j.success);
if !n_times_successful {
return Ok(tx);
return Ok((skip_downstream_error_handlers, tx));
}
let failed_job = past_jobs[past_jobs.len() - 1].clone();
@@ -905,6 +926,7 @@ async fn apply_schedule_handlers<
let on_recovery_result = handle_on_recovery(
db,
tx,
job_id,
schedule_path,
script_path,
schedule.is_flow,
@@ -915,9 +937,6 @@ async fn apply_schedule_handlers<
times,
started_at,
schedule.on_recovery_extra_args,
&schedule.email,
&schedule_to_user(&schedule.path),
username_to_permissioned_as(&schedule.edited_by),
)
.await;
@@ -946,7 +965,7 @@ async fn apply_schedule_handlers<
}
}
Ok(tx)
Ok((skip_downstream_error_handlers, tx))
}
pub async fn handle_on_failure<
@@ -957,6 +976,7 @@ pub async fn handle_on_failure<
>(
db: &Pool<Postgres>,
tx: QueueTransaction<'c, R>,
job_id: Uuid,
schedule_path: &str,
script_path: &str,
is_flow: bool,
@@ -966,18 +986,19 @@ pub async fn handle_on_failure<
failed_times: i32,
started_at: DateTime<Utc>,
extra_args: Option<serde_json::Value>,
username: &str,
email: &str,
permissioned_as: String,
priority: Option<i16>,
) -> windmill_common::error::Result<(Uuid, QueueTransaction<'c, R>)> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_failure_path, db, w_id).await?;
let mut extra = HashMap::new();
extra.insert("schedule_path".to_string(), to_raw_value(&schedule_path));
extra.insert("workspace_id".to_string(), to_raw_value(&w_id));
extra.insert("job_id".to_string(), to_raw_value(&job_id));
extra.insert("path".to_string(), to_raw_value(&script_path));
extra.insert("is_flow".to_string(), to_raw_value(&is_flow));
extra.insert("started_at".to_string(), to_raw_value(&started_at));
extra.insert("email".to_string(), to_raw_value(&email));
extra.insert("failed_times".to_string(), to_raw_value(&failed_times));
if let Some(args_v) = extra_args {
@@ -992,6 +1013,16 @@ pub async fn handle_on_failure<
}
}
// TODO: This should be inject when the EH is defined in the FE.
if on_failure_path
.to_string()
.eq("script/hub/5792/workspace-or-schedule-error-handler-slack")
{
// default slack error handler being used -> we need to inject the slack token
let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}");
extra.insert("slack".to_string(), to_raw_value(&slack_resource));
}
let tx = PushIsolationLevel::Transaction(tx);
let (uuid, tx) = push(
&db,
@@ -999,13 +1030,13 @@ pub async fn handle_on_failure<
w_id,
payload,
PushArgs { extra, args: result.to_owned() },
username,
email,
permissioned_as,
None,
None,
ERROR_HANDLER_USERNAME,
ERROR_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
None,
None,
Some(job_id),
Some(job_id),
None,
false,
false,
@@ -1041,6 +1072,7 @@ async fn handle_on_recovery<
>(
db: &Pool<Postgres>,
tx: QueueTransaction<'c, R>,
job_id: Uuid,
schedule_path: &str,
script_path: &str,
is_flow: bool,
@@ -1051,9 +1083,6 @@ async fn handle_on_recovery<
successful_times: i32,
successful_job_started_at: DateTime<Utc>,
extra_args: Option<serde_json::Value>,
username: &str,
email: &str,
permissioned_as: String,
) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
let (payload, tag) = get_payload_tag_from_prefixed_path(on_recovery_path, db, w_id).await?;
@@ -1086,6 +1115,15 @@ async fn handle_on_recovery<
));
}
}
// TODO: This should be inject when the EH is defined in the FE.
if on_recovery_path
.to_string()
.eq("script/hub/2430/slack/schedule-recovery-handler-slack")
{
// default slack error handler being used -> we need to inject the slack token
let slack_resource = format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}");
args.insert("slack".to_string(), json!(slack_resource));
}
let tx = PushIsolationLevel::Transaction(tx);
let (uuid, tx) = push(
&db,
@@ -1093,13 +1131,13 @@ async fn handle_on_recovery<
w_id,
payload,
args,
username,
email,
permissioned_as,
None,
None,
ERROR_HANDLER_USERNAME,
ERROR_HANDLER_USER_EMAIL,
ERROR_HANDLER_USER_GROUP.to_string(),
None,
None,
Some(job_id),
Some(job_id),
None,
false,
false,
frontend/src/lib/components/ErrorOrRecoveryHandler.svelte
+5 -2
View File
@@ -21,7 +21,7 @@
export let handlerPath: string | undefined
export let handlerExtraArgs: Record<string, any>
export let customInitialScriptPath: string | undefined
export let customInitialScriptPath: string | undefined
export let customScriptTemplate: string
export let customHandlerKind: 'flow' | 'script' = 'script'
let customHandlerSchema: Schema | undefined
@@ -121,6 +121,7 @@
handlerPath !== slackHandlerScriptPath &&
loadHandlerScriptArgs(handlerPath, [
'path',
'job_id',
'is_flow',
'schedule_path',
'error',
@@ -134,6 +135,7 @@
$: loadHandlerScriptArgs(slackHandlerScriptPath, [
'path',
'job_id',
'is_flow',
'schedule_path',
'error',
@@ -142,7 +144,8 @@
'started_at',
'success_times',
'success_result',
'success_started_at'
'success_started_at',
'slack'
]).then((schema) => (slackHandlerSchema = schema))
</script>
frontend/src/lib/components/ScheduleEditorInner.svelte
+146 -21
View File
@@ -10,14 +10,21 @@
import ErrorOrRecoveryHandler from '$lib/components/ErrorOrRecoveryHandler.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import Tooltip from '$lib/components/Tooltip.svelte'
import { FlowService, ScheduleService, Script, ScriptService, type Flow } from '$lib/gen'
import {
FlowService,
ScheduleService,
Script,
ScriptService,
type Flow,
SettingService
} from '$lib/gen'
import { enterpriseLicense, userStore, workspaceStore } from '$lib/stores'
import { canWrite, emptyString, formatCron, sendUserToast } from '$lib/utils'
import { faList, faSave } from '@fortawesome/free-solid-svg-icons'
import { createEventDispatcher } from 'svelte'
import Section from '$lib/components/Section.svelte'
const slackErrorHandler = 'hub/2431/slack/schedule-error-handler-slack'
const slackErrorHandler = 'hub/5792/workspace-or-schedule-error-handler-slack'
const slackRecoveryHandler = 'hub/2430/slack/schedule-recovery-handler-slack'
let initialPath = ''
@@ -27,6 +34,7 @@
let itemKind: 'flow' | 'script' = 'script'
let errorHandleritemKind: 'flow' | 'script' = 'script'
let wsErrorHandlerMuted: boolean = false
let errorHandlerPath: string | undefined = undefined
let errorHandlerCustomInitialPath: string | undefined = undefined
let errorHandlerSelected: 'custom' | 'slack' = 'slack'
@@ -56,23 +64,66 @@
drawer?.openDrawer()
}
export function openNew(is_flow: boolean, initial_script_path?: string) {
export async function openNew(is_flow: boolean, initial_script_path?: string) {
let defaultErrorHandlerMaybe = undefined
let defaultRecoveryHandlerMaybe = undefined
if ($workspaceStore) {
defaultErrorHandlerMaybe = await SettingService.getGlobal({
key: 'default_error_handler_' + $workspaceStore!
})
defaultRecoveryHandlerMaybe = await SettingService.getGlobal({
key: 'default_recovery_handler_' + $workspaceStore!
})
}
edit = false
itemKind = is_flow ? 'flow' : 'script'
initialScriptPath = initial_script_path ?? ''
path = initialScriptPath
initialPath = initialScriptPath
script_path = initialScriptPath
errorHandlerSelected = $enterpriseLicense ? 'slack' : 'custom'
errorHandleritemKind = 'script'
errorHandlerPath = undefined
errorHandlerCustomInitialPath = undefined
errorHandlerExtraArgs = {}
recoveryHandlerSelected = $enterpriseLicense ? 'slack' : 'custom'
recoveryHandlerPath = undefined
recoveryHandlerCustomInitialPath = undefined
recoveryHandlerItemKind = 'script'
recoveryHandlerExtraArgs = {}
if (defaultErrorHandlerMaybe !== undefined && defaultErrorHandlerMaybe !== null) {
console.log(defaultErrorHandlerMaybe)
wsErrorHandlerMuted = defaultErrorHandlerMaybe['wsErrorHandlerMuted']
let splitted = (defaultErrorHandlerMaybe['errorHandlerPath'] as string).split('/')
errorHandleritemKind = splitted[0] as 'flow' | 'script'
errorHandlerPath = splitted.slice(1)?.join('/')
errorHandlerExtraArgs = defaultErrorHandlerMaybe['errorHandlerExtraArgs']
errorHandlerCustomInitialPath = errorHandlerPath
if (errorHandlerPath === slackErrorHandler) {
errorHandlerSelected = 'slack'
} else {
errorHandlerSelected = 'custom'
}
failedTimes = defaultErrorHandlerMaybe['failedTimes']
failedExact = defaultErrorHandlerMaybe['failedExact']
} else {
wsErrorHandlerMuted = false
errorHandlerPath = undefined
errorHandleritemKind = 'script'
errorHandlerExtraArgs = {}
errorHandlerCustomInitialPath = undefined
errorHandlerSelected = $enterpriseLicense ? 'slack' : 'custom'
}
if (defaultRecoveryHandlerMaybe !== undefined && defaultRecoveryHandlerMaybe !== null) {
let splitted = (defaultRecoveryHandlerMaybe['recoveryHandlerPath'] as string).split('/')
recoveryHandlerItemKind = splitted[0] as 'flow' | 'script'
recoveryHandlerPath = splitted.slice(1)?.join('/')
recoveryHandlerExtraArgs = defaultRecoveryHandlerMaybe['recoveryHandlerExtraArgs']
recoveryHandlerCustomInitialPath = recoveryHandlerPath
if (recoveryHandlerPath === slackRecoveryHandler) {
recoveryHandlerSelected = 'slack'
} else {
recoveryHandlerSelected = 'custom'
}
recoveredTimes = defaultRecoveryHandlerMaybe['recoveredTimes']
} else {
recoveryHandlerPath = undefined
recoveryHandlerItemKind = 'script'
recoveryHandlerExtraArgs = {}
recoveryHandlerCustomInitialPath = undefined
recoveryHandlerSelected = $enterpriseLicense ? 'slack' : 'custom'
}
timezone = Intl.DateTimeFormat().resolvedOptions().timeZone
drawer?.openDrawer()
}
@@ -113,6 +164,48 @@
}
}
async function saveAsDefaultErrorHandler() {
if (!$enterpriseLicense) {
sendUserToast(`Setting default error handler is an enterprise edition feature`, true)
return
}
if ($workspaceStore && errorHandlerPath !== undefined) {
await SettingService.setGlobal({
key: 'default_error_handler_' + $workspaceStore!,
requestBody: {
value: {
wsErrorHandlerMuted: wsErrorHandlerMuted,
errorHandlerPath: `${errorHandleritemKind}/${errorHandlerPath}`,
errorHandlerExtraArgs: errorHandlerExtraArgs,
failedTimes: failedTimes,
failedExact: failedExact
}
}
})
sendUserToast(`Default error handler saved to ${errorHandlerPath}`, false)
}
}
async function saveAsDefaultRecoveryHandler() {
if (!$enterpriseLicense) {
sendUserToast(`Setting default recovery handler is an enterprise edition feature`, true)
return
}
if ($workspaceStore && errorHandlerPath !== undefined) {
await SettingService.setGlobal({
key: 'default_recovery_handler_' + $workspaceStore!,
requestBody: {
value: {
recoveryHandlerPath: `${recoveryHandlerItemKind}/${recoveryHandlerPath}`,
recoveryHandlerExtraArgs: recoveryHandlerExtraArgs,
recoveredTimes: recoveredTimes
}
}
})
sendUserToast(`Default recovery handler saved to ${errorHandlerPath}`, false)
}
}
let can_write = true
async function loadSchedule(): Promise<void> {
try {
@@ -125,6 +218,7 @@
timezone = s.timezone
script_path = s.script_path ?? ''
is_flow = s.is_flow
wsErrorHandlerMuted = s.ws_error_handler_muted ?? false
if (s.on_failure) {
let splitted = s.on_failure.split('/')
errorHandleritemKind = splitted[0] as 'flow' | 'script'
@@ -162,12 +256,6 @@
}
async function scheduleScript(): Promise<void> {
if (errorHandlerSelected === 'slack' && !emptyString(errorHandlerPath)) {
errorHandlerExtraArgs['slack'] = '$res:f/slack_bot/bot_token'
}
if (recoveryHandlerSelected === 'slack' && !emptyString(recoveryHandlerPath)) {
recoveryHandlerExtraArgs['slack'] = '$res:f/slack_bot/bot_token'
}
if (edit) {
await ScheduleService.updateSchedule({
workspace: $workspaceStore!,
@@ -184,7 +272,8 @@
? `${recoveryHandlerItemKind}/${recoveryHandlerPath}`
: undefined,
on_recovery_times: recoveredTimes,
on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {}
on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {},
ws_error_handler_muted: wsErrorHandlerMuted
}
})
sendUserToast(`Schedule ${path} updated`)
@@ -207,7 +296,8 @@
? `${recoveryHandlerItemKind}/${recoveryHandlerPath}`
: undefined,
on_recovery_times: recoveredTimes,
on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {}
on_recovery_extra_args: recoveryHandlerPath ? recoveryHandlerExtraArgs : {},
ws_error_handler_muted: wsErrorHandlerMuted
}
})
sendUserToast(`Schedule ${path} created`)
@@ -338,6 +428,27 @@
</div>
</Section>
<Section label="Error handler">
<svelte:fragment slot="action">
<div class="flex flex-row items-center gap-2">
<Button
disabled={emptyString(errorHandlerPath)}
btnClasses="text-center"
color="light"
size="xs"
startIcon={{ icon: faSave }}
on:click={saveAsDefaultErrorHandler}
>
Save as default
</Button>
</div>
</svelte:fragment>
<div class="flex flex-row">
<Toggle
disabled={!can_write}
bind:checked={wsErrorHandlerMuted}
options={{ right: 'Mute workspace error handler for this schedule' }}
/>
</div>
<ErrorOrRecoveryHandler
isEditable={can_write}
handlersOnlyForEe={['slack']}
@@ -406,6 +517,20 @@
{#if !$enterpriseLicense}<span class="text-normal text-2xs">(ee only)</span>{/if}
</div>
</svelte:fragment>
<svelte:fragment slot="action">
<div class="flex flex-row items-center gap-2">
<Button
disabled={emptyString(recoveryHandlerPath)}
btnClasses="text-center"
color="light"
size="xs"
startIcon={{ icon: faSave }}
on:click={saveAsDefaultRecoveryHandler}
>
Save as default
</Button>
</div>
</svelte:fragment>
<ErrorOrRecoveryHandler
isEditable={can_write && !emptyString($enterpriseLicense)}
frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte
+1 -1
View File
@@ -33,7 +33,7 @@
import Toggle from '$lib/components/Toggle.svelte'
import TestOpenaiKey from '$lib/components/copilot/TestOpenaiKey.svelte'
const slackErrorHandler = 'hub/2431/slack/schedule-error-handler-slack'
const slackErrorHandler = 'hub/5792/workspace-or-schedule-error-handler-slack'
let initialPath: string
let scriptPath: string