feat: min workers in worker group alert + zombie job critical alert (#4307)

* feat: min workers in worker group alert + zombie job critical alert

* updatee ee ref

backend/.sqlx/query-00ce4ed3ca0eac7cb6283b047353a64b9e78c4beb423f04baef9a53fbf87e9f9.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO alerts (alert_type, message) VALUES ('recovered_critical_error', $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "00ce4ed3ca0eac7cb6283b047353a64b9e78c4beb423f04baef9a53fbf87e9f9"
+}

backend/.sqlx/query-050fb876e10ad13654dbbde4532f408ff1ac92ed0f5d31a3ef6c58313e1f8671.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT healthy, created_at FROM healthchecks WHERE check_type = 'min_alive_workers_' || $1 ORDER BY created_at DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "healthy",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "050fb876e10ad13654dbbde4532f408ff1ac92ed0f5d31a3ef6c58313e1f8671"
+}

backend/.sqlx/query-354f88b23d20f92c6b6d5bdd8d6c69b08c6a86116cbfd0ecad8f112f7f49d8d1.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT SUBSTRING(name, 9) as \"name!\", (config.config->'min_alive_workers_alert_threshold')::INT as \"threshold!\" \n            FROM config \n            WHERE name LIKE 'worker__%' AND config->'min_alive_workers_alert_threshold' IS NOT NULL",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "name!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "threshold!",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null
+    ]
+  },
+  "hash": "354f88b23d20f92c6b6d5bdd8d6c69b08c6a86116cbfd0ecad8f112f7f49d8d1"
+}

backend/.sqlx/query-3aa1ca29c751f13ea1249c7d43c8b2ce70503935645973fb6a5b35ec47ad1c4c.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO healthchecks (check_type, healthy) \n                        VALUES ('min_alive_workers_' || $1, false)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "3aa1ca29c751f13ea1249c7d43c8b2ce70503935645973fb6a5b35ec47ad1c4c"
+}

backend/.sqlx/query-4c0067c2135a259aea5cc2db60f7375a9a33671be8ef406427d90f67a98c9c9f.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO alerts (alert_type, message) VALUES ('critical_error', $1)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "4c0067c2135a259aea5cc2db60f7375a9a33671be8ef406427d90f67a98c9c9f"
+}

backend/.sqlx/query-7fa8fca5b5cdf147da0cf8ae5a0d9fdee37dd8df66974c8d33f58a88b1bca537.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM healthchecks WHERE check_type LIKE 'min_alive_workers_%' AND created_at < NOW() - INTERVAL '48 hours'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "7fa8fca5b5cdf147da0cf8ae5a0d9fdee37dd8df66974c8d33f58a88b1bca537"
+}

backend/.sqlx/query-8dd1de2aca8c6c9ffaddd2c41c3a614a50fa5fd03c2d3b9a41bd85a7f156345e.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT count(*) \n                    FROM worker_ping \n                    WHERE worker_group LIKE $1 AND ping_at > now() - INTERVAL '2 minutes'",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "count",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "8dd1de2aca8c6c9ffaddd2c41c3a614a50fa5fd03c2d3b9a41bd85a7f156345e"
+}

backend/.sqlx/query-b140388b2e31d15ab4b3a348b6aeec3da9216d4eee23d81fdf6ade95cb7aa1ef.json

@@ -0,0 +1,14 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO healthchecks (check_type, healthy) VALUES ('min_alive_workers_' || $1, true)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "b140388b2e31d15ab4b3a348b6aeec3da9216d4eee23d81fdf6ade95cb7aa1ef"
+}

backend/.sqlx/query-c1ceb03e0eafd2e494c06cef1ec96f78c2cfd42ed7d1ca6cfd35288b561120db.json

@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT healthy, created_at FROM healthchecks WHERE check_type = 'min_alive_workers_' || $1 AND created_at > NOW() - INTERVAL '24 hours' ORDER BY created_at DESC",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "healthy",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "created_at",
+        "type_info": "Timestamptz"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "c1ceb03e0eafd2e494c06cef1ec96f78c2cfd42ed7d1ca6cfd35288b561120db"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-27b5eece4d6b0a54e9caee0933d29a2f8dc92c57
+b86b52e5d1ed2a67ee89c5924e333a28da8b76ca

backend/migrations/20240829143329_alert_center.down.sql

@@ -0,0 +1,3 @@
+-- Add down migration script here
+drop table alerts;
+drop table healthchecks;

backend/migrations/20240829143329_alert_center.up.sql

@@ -0,0 +1,16 @@
+-- Add up migration script here
+create table alerts (
+  id serial PRIMARY KEY,
+  alert_type varchar(50) NOT NULL,
+  message text NOT NULL,
+  created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+create table healthchecks (
+  id bigserial PRIMARY KEY,
+  check_type varchar(50) NOT NULL,
+  healthy boolean NOT NULL,
+  created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+create index healthchecks_check_type_created_at on healthchecks(check_type, created_at);

backend/src/monitor.rs

@@ -21,6 +21,8 @@ use windmill_api::{
     oauth2_ee::{build_oauth_clients, OAuthClient},
     DEFAULT_BODY_LIMIT, IS_SECURE, OAUTH_CLIENTS, REQUEST_SIZE_LIMIT, SAML_METADATA, SCIM_TOKEN,
 };
+#[cfg(feature = "enterprise")]
+use windmill_common::ee::worker_groups_alerts;
 use windmill_common::{
     auth::JWT_SECRET,
     ee::CriticalErrorChannel,
@@ -40,7 +42,7 @@ use windmill_common::{
     oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
     server::load_server_config,
     users::truncate_token,
-    utils::{now_from_db, rd_string},
+    utils::{now_from_db, rd_string, report_critical_error},
     worker::{
         load_worker_config, make_pull_query, make_suspended_pull_query, reload_custom_tags_setting,
         DEFAULT_TAGS_PER_WORKSPACE, DEFAULT_TAGS_WORKSPACES, SERVER_CONFIG, WORKER_CONFIG,
@@ -835,11 +837,19 @@ pub async fn monitor_db(
         }
     };
 
+    let worker_groups_alerts_f = async {
+        #[cfg(feature = "enterprise")]
+        if server_mode {
+            worker_groups_alerts(&db).await;
+        }
+    };
+
     join!(
         expired_items_f,
         zombie_jobs_f,
         expose_queue_metrics_f,
-        verify_license_key_f
+        verify_license_key_f,
+        worker_groups_alerts_f
     );
 }
 
@@ -1051,12 +1061,12 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
         }
         for r in restarted {
-            tracing::error!(
+            let error_message = format!(
                 "Zombie job detected, restarting it: {} {} {:?}",
-                r.id,
-                r.workspace_id,
-                r.last_ping
+                r.id, r.workspace_id, r.last_ping
             );
+            tracing::error!(error_message);
+            report_critical_error(error_message, db.clone()).await;
         }
     }
 
@@ -1159,11 +1169,12 @@ async fn handle_zombie_flows(
                 .get(0)
                 .is_some_and(|x| matches!(x, FlowStatusModule::WaitingForPriorSteps { .. }))
         }) {
-            tracing::error!(
+            let error_message = format!(
                 "Zombie flow detected: {} in workspace {}. It hasn't started yet, restarting it.",
-                flow.id,
-                flow.workspace_id
+                flow.id, flow.workspace_id
             );
+            tracing::error!(error_message);
+            report_critical_error(error_message, db.clone()).await;
             // if the flow hasn't started and is a zombie, we can simply restart it
             sqlx::query!(
                 "UPDATE queue SET running = false, started_at = null WHERE id = $1 AND canceled = false",
@@ -1183,6 +1194,7 @@ async fn handle_zombie_flows(
                     format!("Flow {id} was cancelled because it")
                 }
             );
+            report_critical_error(reason.clone(), db.clone()).await;
             cancel_zombie_flow_job(db, flow, &rsmq, reason).await?;
         }
     }

backend/windmill-common/src/ee.rs

@@ -1,3 +1,4 @@
+use crate::db::DB;
 use crate::ee::LicensePlan::Community;
 #[cfg(feature = "enterprise")]
 use crate::error;
@@ -26,7 +27,10 @@ pub async fn get_license_plan() -> LicensePlan {
 #[serde(untagged)]
 pub enum CriticalErrorChannel {}
 
-pub async fn trigger_critical_error_channels(_error_message: String) {}
+#[cfg(feature = "enterprise")]
+pub async fn send_critical_error(_error_message: String) {}
+#[cfg(feature = "enterprise")]
+pub async fn send_recovered_critical_error(_error_message: String) {}
 
 #[cfg(feature = "enterprise")]
 pub async fn schedule_key_renewal(_http_client: &reqwest::Client, _db: &crate::db::DB) -> () {
@@ -51,3 +55,6 @@ pub async fn create_customer_portal_session(
     // Implementation is not open source
     Ok("".to_string())
 }
+
+#[cfg(feature = "enterprise")]
+pub async fn worker_groups_alerts(_db: &DB) {}

backend/windmill-common/src/utils.rs

@@ -6,9 +6,9 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
-#[cfg(feature = "enterprise")]
-use crate::ee::trigger_critical_error_channels;
 use crate::ee::LICENSE_KEY_ID;
+#[cfg(feature = "enterprise")]
+use crate::ee::{send_critical_error, send_recovered_critical_error};
 use crate::error::{to_anyhow, Error, Result};
 use crate::global_settings::UNIQUE_ID_SETTING;
 use crate::server::Smtp;
@@ -245,8 +245,35 @@ pub async fn send_email(
     return Ok(());
 }
 
-pub async fn report_critical_error(error_message: String) -> () {
+pub async fn report_critical_error(error_message: String, _db: DB) -> () {
     tracing::error!("CRITICAL ERROR: {error_message}");
+
+    if let Err(err) = sqlx::query!(
+        "INSERT INTO alerts (alert_type, message) VALUES ('critical_error', $1)",
+        error_message
+    )
+    .execute(&_db)
+    .await
+    {
+        tracing::error!("Failed to save critical error to database: {}", err);
+    }
+
     #[cfg(feature = "enterprise")]
-    trigger_critical_error_channels(error_message).await;
+    send_critical_error(error_message).await;
+}
+
+pub async fn report_recovered_critical_error(message: String, _db: DB) -> () {
+    tracing::info!("RECOVERED CRITICAL ERROR: {message}");
+
+    if let Err(err) = sqlx::query!(
+        "INSERT INTO alerts (alert_type, message) VALUES ('recovered_critical_error', $1)",
+        message
+    )
+    .execute(&_db)
+    .await
+    {
+        tracing::error!("Failed to save critical error to database: {}", err);
+    }
+    #[cfg(feature = "enterprise")]
+    send_recovered_critical_error(message).await;
 }

backend/windmill-queue/src/jobs.rs

@@ -886,16 +886,19 @@ pub async fn add_completed_job<
         if queued_job.email == ERROR_HANDLER_USER_EMAIL {
             let base_url = BASE_URL.read().await;
             let w_id = &queued_job.workspace_id;
-            report_critical_error(format!(
-                "Workspace error handler job failed ({base_url}/run/{}?workspace={w_id}){}",
-                queued_job.id,
-                queued_job
-                    .parent_job
-                    .map(|id| format!(
-                        " trying to handle failed job ({base_url}/run/{id}?workspace={w_id})"
-                    ))
-                    .unwrap_or("".to_string()),
-            ))
+            report_critical_error(
+                format!(
+                    "Workspace error handler job failed ({base_url}/run/{}?workspace={w_id}){}",
+                    queued_job.id,
+                    queued_job
+                        .parent_job
+                        .map(|id| format!(
+                            " trying to handle failed job ({base_url}/run/{id}?workspace={w_id})"
+                        ))
+                        .unwrap_or("".to_string()),
+                ),
+                db.clone(),
+            )
             .await;
         } else if queued_job.email == SCHEDULE_ERROR_HANDLER_USER_EMAIL {
             let base_url = BASE_URL.read().await;
@@ -961,7 +964,7 @@ pub async fn add_completed_job<
                             "Could not push workspace error handler for failed job ({base_url}/run/{}?workspace={w_id}): {}",
                             queued_job.id,
                             err
-                        ))
+                        ), db.clone())
                         .await;
                     }
                 }
@@ -1143,10 +1146,10 @@ pub async fn report_error_to_workspace_handler_or_critical_side_channel<
                 queued_job.id,
                 err
             );
-            report_critical_error(error_message).await;
+            report_critical_error(error_message, db.clone()).await;
         }
     } else {
-        report_critical_error(error_message).await;
+        report_critical_error(error_message, db.clone()).await;
     }
 }
 

frontend/src/lib/components/WorkspaceGroup.svelte

@@ -17,6 +17,7 @@
 	import Label from './Label.svelte'
 	import AutoComplete from 'simple-svelte-autocomplete'
 	import YAML from 'yaml'
+	import Toggle from './Toggle.svelte'
 
 	export let name: string
 	export let config:
@@ -29,6 +30,7 @@
 				init_bash?: string
 				additional_python_paths?: string[]
 				pip_local_dependencies?: string[]
+				min_alive_workers_alert_threshold?: number
 		  }
 	export let activeWorkers: number
 	export let customTags: string[] | undefined
@@ -62,6 +64,7 @@
 		env_vars_allowlist?: string[]
 		additional_python_paths?: string[]
 		pip_local_dependencies?: string[]
+		min_alive_workers_alert_threshold?: number
 	} = {}
 
 	function loadNConfig() {
@@ -465,6 +468,42 @@
 					{/if}
 				{/if}
 			</Section>
+			{#if nconfig !== undefined}
+				<div class="mt-8" />
+				<Section label="Alerts" tooltip="Alert is sent to the configured critical error channels">
+					<Toggle
+						size="sm"
+						options={{
+							right: 'Send an alert when the number of alive workers falls below a given threshold'
+						}}
+						checked={nconfig?.min_alive_workers_alert_threshold !== undefined ?? false}
+						on:change={(ev) => {
+							if (nconfig !== undefined) {
+								nconfig.min_alive_workers_alert_threshold = ev.detail ? 1 : undefined
+								dirty = true
+							}
+						}}
+						disabled{!$enterpriseLicense}
+					/>
+					{#if nconfig.min_alive_workers_alert_threshold !== undefined}
+						<div class="flex flex-row items-center justify-between">
+							<div class="flex flex-row items-center text-sm gap-2">
+								<p>Triggered when number of workers in group is lower than</p>
+								<input
+									type="number"
+									class="!w-14 text-center"
+									disabled={!$enterpriseLicense}
+									min="1"
+									bind:value={nconfig.min_alive_workers_alert_threshold}
+									on:change={(ev) => {
+										dirty = true
+									}}
+								/>
+							</div>
+						</div>
+					{/if}
+				</Section>
+			{/if}
 		{:else if selected == 'dedicated'}
 			{#if nconfig?.dedicated_worker != undefined}
 				<input
@@ -789,6 +828,13 @@
 						variant="contained"
 						color="dark"
 						on:click={async () => {
+							if (
+								nconfig?.min_alive_workers_alert_threshold &&
+								nconfig?.min_alive_workers_alert_threshold < 1
+							) {
+								sendUserToast('Minimum alive workers alert threshold must be at least 1', true)
+								return
+							}
 							customEnvVars.forEach((envvar) => {
 								if (
 									nconfig.env_vars_static !== undefined &&

frontend/src/lib/components/instanceSettings.ts

@@ -120,9 +120,9 @@ export const settings: Record<string, Setting[]> = {
 			ee_only: ''
 		},
 		{
-			label: 'Critical Error Channels',
+			label: 'Critical Alert Channels',
 			description:
-				'Channels to send critical errors to. SMTP must be configured for the email channel.',
+				'Channels to send critical alerts to. SMTP must be configured for the email channel.',
 			key: 'critical_error_channels',
 			fieldType: 'critical_error_channels',
 			storage: 'setting',