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

feat: critical error side channel (#3625)

Browse Source 
* feat: critical error side channel

* feat: trigger workspace error handler or critical channel on schedule error handling failure

* fix: build

* feat: finish critical error side channel

* feat: channels instead of just emails + move to EE

* fix: remove superfluous func

* fix: nits + ee ref

* fix: open source build

* fix: nits

* chore: update ee ref
This commit is contained in:
HugoCasa
2024-05-02 20:21:23 +02:00
committed by GitHub
parent cce104f80e
commit cb132671d3
21 changed files with 676 additions and 456 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+1 -1
View File
@@ -9683,7 +9683,6 @@ dependencies = [
"itertools 0.12.1",
"lazy_static",
"magic-crypt",
"mail-send",
"mime_guess",
"object_store",
"openidconnect",
@@ -9777,6 +9776,7 @@ dependencies = [
"itertools 0.12.1",
"lazy_static",
"magic-crypt",
"mail-send",
"object_store",
"prometheus",
"rand 0.8.5",
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
66d9cbb158ab9a5869a45ba253bf57f2cbb5ecb6
f59f9c1e55e5e93f9eb6c081847cc566611029d2
backend/src/main.rs
+15 -9
View File
@@ -18,11 +18,11 @@ use tokio::fs::DirBuilder;
use windmill_api::HTTP_CLIENT;
use windmill_common::{
global_settings::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CUSTOM_TAGS_SETTING,
DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS, EXPOSE_DEBUG_METRICS_SETTING,
EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
CUSTOM_TAGS_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING, ENV_SETTINGS,
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
@@ -47,10 +47,11 @@ use windmill_worker::{
use crate::monitor::{
initial_load, load_keep_job_dir, load_require_preexisting_user, load_tag_per_workspace_enabled,
monitor_db, monitor_pool, reload_base_url_setting, reload_bunfig_install_scopes_setting,
reload_extra_pip_index_url_setting, reload_hub_base_url_setting,
reload_job_default_timeout_setting, reload_license_key, reload_npm_config_registry_setting,
reload_pip_index_url_setting, reload_retention_period_setting, reload_scim_token_setting,
reload_server_config, reload_worker_config,
reload_critical_error_channels_setting, reload_extra_pip_index_url_setting,
reload_hub_base_url_setting, reload_job_default_timeout_setting, reload_license_key,
reload_npm_config_registry_setting, reload_pip_index_url_setting,
reload_retention_period_setting, reload_scim_token_setting, reload_server_config,
reload_worker_config,
};
#[cfg(feature = "parquet")]
@@ -503,6 +504,11 @@ Windmill Community Edition {GIT_VERSION}
tracing::error!(error = %e, "Could not reload hub base url setting");
}
},
CRITICAL_ERROR_CHANNELS_SETTING => {
if let Err(e) = reload_critical_error_channels_setting(&db).await {
tracing::error!(error = %e, "Could not reload critical error emails setting");
}
},
a @_ => {
tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
}
backend/src/monitor.rs
+38 -7
View File
@@ -22,15 +22,17 @@ use windmill_api::{
DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN,
};
use windmill_common::{
ee::CriticalErrorChannel,
error,
flow_status::FlowStatusModule,
global_settings::{
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, DEFAULT_TAGS_PER_WORKSPACE_SETTING,
EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING,
HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, PIP_INDEX_URL_SETTING,
REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING,
RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ERROR_CHANNELS_SETTING,
DEFAULT_TAGS_PER_WORKSPACE_SETTING, EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING,
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING,
},
jobs::QueuedJob,
oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
@@ -40,7 +42,8 @@ use windmill_common::{
load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE, SERVER_CONFIG,
WORKER_CONFIG,
},
BASE_URL, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL,
METRICS_DEBUG_ENABLED, METRICS_ENABLED,
};
use windmill_queue::cancel_job;
use windmill_worker::{
@@ -143,6 +146,10 @@ pub async fn initial_load(
tracing::error!("Error reloading hub base url: {:?}", e)
}
if let Err(e) = reload_critical_error_channels_setting(&db).await {
tracing::error!("Could not reload critical error emails setting: {:?}", e);
}
#[cfg(feature = "parquet")]
if !_is_agent {
reload_s3_cache_setting(&db).await;
@@ -1074,3 +1081,27 @@ pub async fn reload_hub_base_url_setting(db: &DB, server_mode: bool) -> error::R
Ok(())
}
pub async fn reload_critical_error_channels_setting(db: &DB) -> error::Result<()> {
let critical_error_channels =
load_value_from_global_settings(db, CRITICAL_ERROR_CHANNELS_SETTING).await?;
let critical_error_channels = if let Some(q) = critical_error_channels {
if let Ok(v) = serde_json::from_value::<Vec<CriticalErrorChannel>>(q.clone()) {
v
} else {
tracing::error!(
"Could not parse critical_error_emails setting as an array of channels, found: {:#?}",
&q
);
vec![]
}
} else {
vec![]
};
let mut l = CRITICAL_ERROR_CHANNELS.write().await;
*l = critical_error_channels;
Ok(())
}
backend/windmill-api/Cargo.toml
-1
View File
@@ -72,7 +72,6 @@ async_zip.workspace = true
rsmq_async.workspace = true
regex.workspace = true
bytes.workspace = true
mail-send.workspace = true
samael = { workspace = true, optional = true }
async-recursion.workspace = true
rsa.workspace = true
backend/windmill-api/src/flows.rs
+1 -1
View File
@@ -543,7 +543,7 @@ async fn update_flow(
clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, schedule).await?;
tx = push_scheduled_job(&db, tx, &schedule).await?;
}
}
backend/windmill-api/src/schedule.rs
+4 -4
View File
@@ -221,7 +221,7 @@ async fn create_schedule(
.await?;
if ns.enabled.unwrap_or(true) {
tx = push_scheduled_job(&db, tx, schedule).await?
tx = push_scheduled_job(&db, tx, &schedule).await?
}
tx.commit().await?;
@@ -303,7 +303,7 @@ async fn edit_schedule(
.await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, schedule).await?;
tx = push_scheduled_job(&db, tx, &schedule).await?;
}
tx.commit().await?;
@@ -512,7 +512,7 @@ pub async fn set_enabled(
.await?;
if payload.enabled {
tx = push_scheduled_job(&db, tx, schedule).await?;
tx = push_scheduled_job(&db, tx, &schedule).await?;
}
tx.commit().await?;
@@ -560,7 +560,7 @@ pub async fn set_enabled(
// .await?;
// if payload.enabled {
// tx = push_scheduled_job(&db, tx, schedule).await?;
// tx = push_scheduled_job(&db, tx, &schedule).await?;
// }
// tx.commit().await?;
backend/windmill-api/src/scripts.rs
+1 -1
View File
@@ -554,7 +554,7 @@ async fn create_script(
clear_schedule(tx.transaction_mut(), &schedule.path, &w_id).await?;
if schedule.enabled {
tx = push_scheduled_job(&db, tx, schedule).await?;
tx = push_scheduled_job(&db, tx, &schedule).await?;
}
}
} else {
backend/windmill-api/src/settings.rs
+16 -31
View File
@@ -21,15 +21,17 @@ use axum::{
Json, Router,
};
use mail_send::{mail_builder::MessageBuilder, SmtpClientBuilder};
use serde::Deserialize;
use tokio::time::timeout;
use windmill_common::{
error::{self, to_anyhow, JsonResult, Result},
error::{self, JsonResult, Result},
global_settings::{AUTOMATE_USERNAME_CREATION_SETTING, ENV_SETTINGS, HUB_BASE_URL_SETTING},
server::Smtp,
utils::send_email,
};
#[cfg(feature = "parquet")]
use windmill_common::error::to_anyhow;
pub fn global_service() -> Router {
#[warn(unused_mut)]
let r = Router::new()
@@ -67,34 +69,17 @@ pub async fn test_email(
require_super_admin(&db, &authed.email).await?;
let smtp = test_email.smtp;
let to = test_email.to;
let mut client = SmtpClientBuilder::new(smtp.host, smtp.port)
.implicit_tls(smtp.tls_implicit.unwrap_or(false));
if std::env::var("ACCEPT_INVALID_CERTS").is_ok() {
client = client.allow_invalid_certs();
}
let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) {
if !username.is_empty() {
client.credentials((username, password))
} else {
client
}
} else {
client
};
let message = MessageBuilder::new()
.from(("Windmill", smtp.from.as_str()))
.to(to.clone())
.subject("Test email from Windmill")
.text_body("Test email content");
let dur = Duration::from_secs(3);
timeout(dur, client.connect())
.await
.map_err(to_anyhow)?
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
tracing::info!("Sent test email to {to}");
let client_timeout = Duration::from_secs(3);
send_email(
"Test email from Windmill",
"Test email content",
vec![to],
smtp,
Some(client_timeout),
)
.await?;
Ok("Sent test email".to_string())
}
backend/windmill-api/src/workspaces.rs
+8 -9
View File
@@ -53,7 +53,6 @@ use windmill_common::{
variables::ExportableListableVariable,
};
use windmill_git_sync::handle_deployment_metadata;
use windmill_queue::QueueTransaction;
use crate::oauth2_ee::InstanceEvent;
use crate::variables::{decrypt, encrypt};
@@ -488,25 +487,25 @@ async fn run_slack_message_test_job(
json!(format!("$res:{WORKSPACE_SLACK_BOT_TOKEN_PATH}")),
);
let tx: QueueTransaction<'_, _> = (rsmq.clone(), db.begin().await?).into();
let (uuid, tx) = windmill_queue::handle_on_failure(
let uuid = windmill_queue::push_error_handler(
&db,
tx,
rsmq,
Uuid::parse_str("00000000-0000-0000-0000-000000000000")?,
"slack_message_test",
"slack_message_test",
None,
Some("slack_message_test".to_string()),
false,
w_id.as_str(),
&format!("script/{}", req.hub_script_path.as_str()),
sqlx::types::Json(&fake_result),
0,
Utc::now(),
None,
Some(Utc::now()),
Some(json!(extra_args)),
authed.email.as_str(),
false,
false,
None, // Note: we could mark it as high priority to return result quickly to the user
)
.await?;
tx.commit().await?;
Ok(Json(RunSlackMessageTestJobResponse {
job_uuid: uuid.to_string(),
backend/windmill-common/Cargo.toml
+2 -1
View File
@@ -46,4 +46,5 @@ object_store = { workspace = true, optional = true }
prometheus = { workspace = true, optional = true }
aws-config = { workspace = true, optional = true }
aws-sdk-sts = { workspace = true, optional = true }
indexmap.workspace = true
indexmap.workspace = true
mail-send.workspace = true
backend/windmill-common/src/ee.rs
+7
View File
@@ -1,4 +1,5 @@
use crate::ee::LicensePlan::Community;
use serde::Deserialize;
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -18,3 +19,9 @@ pub async fn get_license_plan() -> LicensePlan {
// Implementation is not open source
return Community;
}
#[derive(Deserialize)]
#[serde(untagged)]
pub enum CriticalErrorChannel {}
pub async fn trigger_critical_error_channels(_error_message: String) {}
backend/windmill-common/src/error.rs
+7 -7
View File
@@ -8,10 +8,7 @@
use axum::body::Body;
use axum::response::Response;
use axum::{
response::IntoResponse,
response::Json,
};
use axum::{response::IntoResponse, response::Json};
use hyper::StatusCode;
use sqlx::migrate::MigrateError;
@@ -45,6 +42,8 @@ pub enum Error {
SqlErr(#[from] sqlx::Error),
#[error("Bad request: {0}")]
BadRequest(String),
#[error("Quota exceeded: {0}")]
QuotaExceeded(String),
#[error("Internal: {0}")]
InternalErr(String),
#[error("Hexadecimal decoding error: {0}")]
@@ -81,9 +80,10 @@ impl IntoResponse for Error {
Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND,
Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN,
Self::SqlErr(_) | Self::BadRequest(_) | Self::OpenAIError(_) => {
axum::http::StatusCode::BAD_REQUEST
}
Self::SqlErr(_)
| Self::BadRequest(_)
| Self::OpenAIError(_)
| Self::QuotaExceeded(_) => axum::http::StatusCode::BAD_REQUEST,
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};
backend/windmill-common/src/global_settings.rs
+1
View File
@@ -24,6 +24,7 @@ pub const OBJECT_STORE_CACHE_CONFIG_SETTING: &str = "object_store_cache_config";
pub const AUTOMATE_USERNAME_CREATION_SETTING: &str = "automate_username_creation";
pub const HUB_BASE_URL_SETTING: &str = "hub_base_url";
pub const CRITICAL_ERROR_CHANNELS_SETTING: &str = "critical_error_channels";
pub const ENV_SETTINGS: [&str; 50] = [
"DISABLE_NSJAIL",
backend/windmill-common/src/lib.rs
+4
View File
@@ -11,6 +11,7 @@ use std::{
sync::{atomic::AtomicBool, Arc},
};
use ee::CriticalErrorChannel;
use error::Error;
use scripts::ScriptLang;
use sqlx::{Pool, Postgres};
@@ -73,6 +74,9 @@ lazy_static::lazy_static! {
pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
pub static ref HUB_BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new(DEFAULT_HUB_BASE_URL.to_string()));
pub static ref CRITICAL_ERROR_CHANNELS: Arc<RwLock<Vec<CriticalErrorChannel>>> = Arc::new(RwLock::new(vec![]));
}
pub async fn shutdown_signal(
backend/windmill-common/src/utils.rs
+71 -3
View File
@@ -6,11 +6,14 @@
* LICENSE-AGPL for a copy of the license.
*/
use crate::ee::LICENSE_KEY_ID;
use crate::ee::{trigger_critical_error_channels, LICENSE_KEY_ID};
use crate::error::{to_anyhow, Error, Result};
use crate::global_settings::UNIQUE_ID_SETTING;
use crate::server::Smtp;
use crate::DB;
use git_version::git_version;
use mail_send::mail_builder::MessageBuilder;
use mail_send::SmtpClientBuilder;
use rand::{distributions::Alphanumeric, thread_rng, Rng};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
@@ -85,12 +88,20 @@ pub async fn query_elems_from_hub(
url: &str,
query_params: Option<Vec<(&str, String)>>,
db: &DB,
) -> Result<(reqwest::StatusCode, reqwest::header::HeaderMap, axum::body::Body)> {
) -> Result<(
reqwest::StatusCode,
reqwest::header::HeaderMap,
axum::body::Body,
)> {
let response = http_get_from_hub(http_client, url, false, query_params, db).await?;
let status = response.status();
Ok((status, response.headers().clone(), axum::body::Body::from_stream(response.bytes_stream())))
Ok((
status,
response.headers().clone(),
axum::body::Body::from_stream(response.bytes_stream()),
))
}
pub async fn http_get_from_hub(
@@ -167,3 +178,60 @@ pub enum Mode {
Server,
Standalone,
}
pub async fn send_email(
subject: &str,
content: &str,
to: Vec<String>,
smtp: Smtp,
client_timeout: Option<tokio::time::Duration>,
) -> Result<()> {
let mut client = SmtpClientBuilder::new(smtp.host, smtp.port)
.implicit_tls(smtp.tls_implicit.unwrap_or(false));
if std::env::var("ACCEPT_INVALID_CERTS").is_ok() {
client = client.allow_invalid_certs();
}
let client = if let (Some(username), Some(password)) = (smtp.username, smtp.password) {
if !username.is_empty() {
client.credentials((username, password))
} else {
client
}
} else {
client
};
let message = MessageBuilder::new()
.from(("Windmill", smtp.from.as_str()))
.to(to.clone())
.subject(subject)
.text_body(content);
match client_timeout {
Some(timeout) => {
tokio::time::timeout(timeout, client.connect())
.await
.map_err(to_anyhow)?
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
}
None => {
client
.connect()
.await
.map_err(to_anyhow)?
.send(message)
.await
.map_err(to_anyhow)?;
}
}
tracing::info!("Sent email to {:#?}: {subject}", to);
return Ok(());
}
pub async fn report_critical_error(error_message: String) -> () {
tracing::error!("CRITICAL ERROR: {error_message}");
trigger_critical_error_channels(error_message).await;
}
backend/windmill-queue/src/jobs.rs
+374 -353
View File
File diff suppressed because it is too large Load Diff
backend/windmill-queue/src/schedule.rs
+10 -10
View File
@@ -26,9 +26,9 @@ use windmill_common::{
pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
db: &DB,
mut tx: QueueTransaction<'c, R>,
schedule: Schedule,
schedule: &Schedule,
) -> Result<QueueTransaction<'c, R>> {
let sched = cron::Schedule::from_str(&schedule.schedule)
let sched = cron::Schedule::from_str(schedule.schedule.as_ref())
.map_err(|e| error::Error::BadRequest(e.to_string()))?;
let tz = chrono_tz::Tz::from_str(&schedule.timezone)
@@ -68,9 +68,9 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
let mut args: serde_json::Map<String, serde_json::Value> = serde_json::Map::new();
if let Some(args_v) = schedule.args {
if let Some(args_v) = &schedule.args {
if let serde_json::Value::Object(args_m) = args_v {
args = args_m
args = args_m.clone()
} else {
return Err(error::Error::ExecutionErr(
"args of scripts needs to be dict".to_string(),
@@ -90,7 +90,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
.map(|x| (x.tag, x.dedicated_worker))
.unwrap_or_else(|| (None, None));
(
JobPayload::Flow { path: schedule.script_path, dedicated_worker },
JobPayload::Flow { path: schedule.script_path.clone(), dedicated_worker },
tag,
None,
)
@@ -113,8 +113,8 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
.await?;
if schedule.retry.is_some() {
let parsed_retry =
serde_json::from_value::<Retry>(schedule.retry.unwrap()).map_err(|err| {
let parsed_retry = serde_json::from_value::<Retry>(schedule.retry.clone().unwrap())
.map_err(|err| {
error::Error::InternalErr(format!(
"Unable to parse retry information from schedule: {}",
err.to_string(),
@@ -127,7 +127,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
// if retry is set, we wrap the script into a one step flow with a retry on the module
(
JobPayload::SingleScriptFlow {
path: schedule.script_path,
path: schedule.script_path.clone(),
hash: hash,
retry: parsed_retry,
args: static_args,
@@ -144,7 +144,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
(
JobPayload::ScriptHash {
hash,
path: schedule.script_path,
path: schedule.script_path.clone(),
concurrent_limit: concurrent_limit,
concurrency_time_window_s: concurrency_time_window_s,
cache_ttl: cache_ttl,
@@ -153,7 +153,7 @@ pub async fn push_scheduled_job<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
priority,
},
if schedule.tag.as_ref().is_some_and(|x| x != "") {
schedule.tag
schedule.tag.clone()
} else {
tag
},
backend/windmill-worker/src/worker_flow.rs
+31 -16
View File
@@ -42,6 +42,7 @@ use windmill_common::{
},
flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Suspend},
};
use windmill_queue::schedule::get_schedule_opt;
use windmill_queue::{
add_completed_job, add_completed_job_error, append_logs, get_queued_job,
handle_maybe_scheduled_job, CanceledBy, PushIsolationLevel, WrappedError,
@@ -1201,23 +1202,37 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
&& flow_job.script_path.is_some()
&& status.step == 0
{
let tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
let mut tx: QueueTransaction<'_, R> = (rsmq.clone(), db.begin().await?).into();
match handle_maybe_scheduled_job(
tx,
db,
flow_job.schedule_path.as_ref().unwrap(),
flow_job.script_path.as_ref().unwrap(),
&flow_job.workspace_id,
)
.await
{
Ok(tx) => {
tx.commit().await?;
}
Err(e) => {
tracing::error!("Error during handle_maybe_scheduled_job: {e}");
}
let schedule_path = flow_job.schedule_path.as_ref().unwrap();
let schedule =
get_schedule_opt(tx.transaction_mut(), &flow_job.workspace_id, schedule_path).await?;
tx.commit().await?;
if let Some(schedule) = schedule {
if let Err(err) = handle_maybe_scheduled_job(
rsmq.clone(),
db,
flow_job,
&schedule,
flow_job.script_path.as_ref().unwrap(),
&flow_job.workspace_id,
)
.await
{
match err {
Error::QuotaExceeded(_) => return Err(err.into()),
// scheduling next job failed and could not disable schedule => make zombie job to retry
_ => return Ok(()),
}
};
} else {
tracing::error!(
"Schedule {schedule_path} in {} not found. Impossible to schedule again",
&flow_job.workspace_id
);
}
}
frontend/src/lib/components/InstanceSettings.svelte
+74 -1
View File
@@ -16,7 +16,7 @@
import { capitalize } from '$lib/utils'
import { enterpriseLicense } from '$lib/stores'
import CustomOauth from './CustomOauth.svelte'
import { AlertTriangle, Plus } from 'lucide-svelte'
import { AlertTriangle, Plus, X } from 'lucide-svelte'
import CustomSso from './CustomSso.svelte'
import AuthentikSetting from '$lib/components/AuthentikSetting.svelte'
import AutheliaSetting from '$lib/components/AutheliaSetting.svelte'
@@ -24,6 +24,7 @@
import ZitadelSetting from '$lib/components/ZitadelSetting.svelte'
import Password from './Password.svelte'
import ObjectStoreConfigSettings from './ObjectStoreConfigSettings.svelte'
import { fade } from 'svelte/transition'
export let tab: string = 'Core'
export let hideTabs: boolean = false
@@ -498,6 +499,78 @@
placeholder={setting.placeholder}
bind:value={values[setting.key]}
/>
{:else if setting.fieldType == 'critical_error_channels'}
<div class="w-full">
<div class="flex max-w-md mt-1 gap-2 w-full items-center">
<select disabled>
<option>Tracing</option>
</select>
<input disabled />
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
aria-label="Clear"
disabled
>
<X size={14} />
</button>
</div>
{#if $enterpriseLicense && Array.isArray(values[setting.key])}
{#each values[setting.key] ?? [] as v, i}
<div class="flex max-w-md mt-1 gap-2 w-full items-center">
<select>
<option value="email">Email</option>
</select>
<input
type="email"
placeholder="Email address"
on:input={(e) => {
if (e.target?.['value']) {
values[setting.key][i] = {
email: e.target['value']
}
}
}}
value={v?.email ?? ''}
/>
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
aria-label="Clear"
on:click={() => {
values[setting.key] = values[setting.key].filter(
(_, index) => index !== i
)
}}
>
<X size={14} />
</button>
</div>
{/each}
{/if}
</div>
<div class="flex mt-2 gap-20 items-baseline">
<Button
variant="border"
color="light"
size="xs"
btnClasses="mt-1"
on:click={() => {
if (
values[setting.key] == undefined ||
!Array.isArray(values[setting.key])
) {
values[setting.key] = []
}
values[setting.key] = values[setting.key].concat('')
}}
id="arg-input-add-item"
startIcon={{ icon: Plus }}
disabled={!$enterpriseLicense}
>
Add item
</Button>
</div>
{:else if setting.fieldType == 'object_store_config'}
<ObjectStoreConfigSettings bind:bucket_config={values[setting.key]} />
{:else if setting.fieldType == 'number'}
frontend/src/lib/components/instanceSettings.ts
+10
View File
@@ -17,6 +17,7 @@ export interface Setting {
| 'email'
| 'license_key'
| 'object_store_config'
| 'critical_error_channels'
storage: SettingStorage
isValid?: (value: any) => boolean
error?: string
@@ -102,6 +103,15 @@ export const settings: Record<string, Setting[]> = {
storage: 'setting',
ee_only: ''
},
{
label: 'Critical Error Channels',
description:
'Channels to send critical errors to. SMTP must be configured for the email channel.',
key: 'critical_error_channels',
fieldType: 'critical_error_channels',
storage: 'setting',
ee_only: 'Channels other than tracing are only available in the EE version'
},
{
label: 'Azure OpenAI base path',
description: