Feature: Advanced Webhooks (#369)

* introduce advanced webhook plugin - node group changes, node health changes, metrics changes

Author: JannikSt

.env.example

@@ -37,4 +37,8 @@ WORK_VALIDATION_CONTRACT=
 # Example: [{"server_url": "toploc url", "auth_token": "toploc auth token", "file_prefix_filter": "optional prefix filter"}]
 TOPLOC_CONFIGS=
 S3_CREDENTIALS=
-BUCKET_NAME=
+BUCKET_NAME=
+
+# Webhook configurations as JSON array string
+# Example: [{"url": "webhook url", "bearer_token": "webhook bearer token"}]
+WEBHOOK_CONFIGS=

Cargo.lock

@@ -35,6 +35,7 @@ nalgebra = "0.33.2"
 redis = "0.28.1"
 redis-test = "0.8.0"
 stun = "0.7.0"
+mockito = "1.7.0"
 
 [workspace.package]
 version = "0.2.11"

Cargo.toml

@@ -29,3 +29,6 @@ shared = { workspace = true }
 tokio = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
+
+[dev-dependencies]
+mockito = { workspace = true }

crates/orchestrator/Cargo.toml

@@ -21,10 +21,10 @@ ENV S3_CREDENTIALS=""
 ENV BUCKET_NAME=""
 ENV LOG_LEVEL=""
 ENV HOURLY_S3_UPLOAD_LIMIT="2"
-ENV WEBHOOK_URLS=""
 ENV MODE="full"
 ENV NODE_GROUP_CONFIGS=""
 ENV NODE_GROUP_MANAGEMENT_INTERVAL="10"
+ENV WEBHOOK_CONFIGS=""
 
 RUN echo '#!/bin/sh\n\
 exec /usr/local/bin/orchestrator \
@@ -44,7 +44,6 @@ $([ "$DISABLE_EJECTION" = "true" ] && echo "--disable-ejection") \
 $([ ! -z "$BUCKET_NAME" ] && echo "--bucket-name $BUCKET_NAME") \
 $([ ! -z "$LOG_LEVEL" ] && echo "--log-level $LOG_LEVEL") \
 $([ ! -z "$HOURLY_S3_UPLOAD_LIMIT" ] && echo "--hourly-s3-upload-limit $HOURLY_S3_UPLOAD_LIMIT") \
-$([ ! -z "$WEBHOOK_URLS" ] && echo "--webhook-urls $WEBHOOK_URLS") \
 $([ ! -z "$NODE_GROUP_MANAGEMENT_INTERVAL" ] && echo "--node-group-management-interval $NODE_GROUP_MANAGEMENT_INTERVAL") \
 "$@"' > /entrypoint.sh && \
 chmod +x /entrypoint.sh

crates/orchestrator/Dockerfile

@@ -426,6 +426,7 @@ mod tests {
             app_state.redis_store.clone(),
             app_state.store_context.clone(),
             None,
+            None,
         );
 
         let _ = plugin

crates/orchestrator/src/api/routes/storage.rs

@@ -107,6 +107,7 @@ pub async fn create_test_app_state_with_nodegroups() -> Data<AppState> {
         store.clone(),
         store_context.clone(),
         None,
+        None,
     )));
 
     let mock_storage = MockStorageProvider::new();

crates/orchestrator/src/api/tests/helper.rs

@@ -1,7 +1,6 @@
 use anyhow::Result;
 use shared::models::task::Task;
 
-// TODO: Shouldnt this be handled in a tokio trhead?
 pub trait TaskObserver: Send + Sync {
     fn on_task_created(&self, task: &Task) -> Result<()>;
     fn on_task_deleted(&self, task: Option<Task>) -> Result<()>;

crates/orchestrator/src/events/mod.rs

@@ -25,9 +25,11 @@ use log::debug;
 use log::error;
 use log::info;
 use log::LevelFilter;
+use metrics::webhook_sender::MetricsWebhookSender;
 use metrics::MetricsContext;
 use plugins::node_groups::NodeGroupConfiguration;
 use plugins::node_groups::NodeGroupsPlugin;
+use plugins::webhook::WebhookConfig;
 use plugins::webhook::WebhookPlugin;
 use plugins::SchedulerPlugin;
 use plugins::StatusUpdatePlugin;
@@ -112,10 +114,6 @@ struct Args {
     #[arg(short = 'l', long, default_value = "info")]
     log_level: String,
 
-    /// Webhook urls (comma-separated string)
-    #[arg(long, default_value = "")]
-    webhook_urls: Option<String>,
-
     /// Node group management interval
     #[arg(long, default_value = "10")]
     node_group_management_interval: u64,
@@ -199,6 +197,40 @@ async fn main() -> Result<()> {
     let mut scheduler_plugins: Vec<Box<dyn SchedulerPlugin>> = Vec::new();
     let mut status_update_plugins: Vec<Box<dyn StatusUpdatePlugin>> = vec![];
     let mut node_groups_plugin: Option<Arc<NodeGroupsPlugin>> = None;
+    let mut webhook_plugins: Vec<WebhookPlugin> = vec![];
+
+    let configs = std::env::var("WEBHOOK_CONFIGS").unwrap_or_default();
+    match serde_json::from_str::<Vec<WebhookConfig>>(&configs) {
+        Ok(configs) if !configs.is_empty() => {
+            for config in configs {
+                let plugin = WebhookPlugin::new(config);
+                let plugin_clone = plugin.clone();
+                webhook_plugins.push(plugin_clone);
+                status_update_plugins.push(Box::new(plugin));
+            }
+        }
+        Ok(_) => {
+            info!("No webhook configurations provided");
+        }
+        Err(e) => {
+            error!("Failed to parse webhook configs from environment: {}", e);
+        }
+    }
+
+    let webhook_sender_store = store_context.clone();
+    let webhook_plugins_clone = webhook_plugins.clone();
+    if !webhook_plugins_clone.is_empty() {
+        tasks.spawn(async move {
+            let mut webhook_sender = MetricsWebhookSender::new(
+                webhook_sender_store.clone(),
+                webhook_plugins_clone.clone(),
+            );
+            if let Err(e) = webhook_sender.run().await {
+                error!("Error running webhook sender: {}", e);
+            }
+            Ok(())
+        });
+    }
 
     // Load node group configurations from environment variable
     if let Ok(configs_json) = std::env::var("NODE_GROUP_CONFIGS") {
@@ -210,13 +242,14 @@ async fn main() -> Result<()> {
                     store.clone(),
                     group_store_context.clone(),
                     Some(node_groups_heartbeats.clone()),
+                    Some(webhook_plugins.clone()),
                 );
                 let status_group_plugin = group_plugin.clone();
                 let group_plugin_for_server = group_plugin.clone();
                 node_groups_plugin = Some(Arc::new(group_plugin_for_server));
                 scheduler_plugins.push(Box::new(group_plugin));
                 status_update_plugins.push(Box::new(status_group_plugin));
-                info!("Node group plugin initialized",);
+                info!("Node group plugin initialized");
             }
             Ok(_) => {
                 info!(
@@ -277,17 +310,6 @@ async fn main() -> Result<()> {
         let status_update_store_context = store_context.clone();
         let status_update_heartbeats = heartbeats.clone();
         let status_update_contracts = contracts.clone();
-        let webhook_urls: Vec<String> = args
-            .webhook_urls
-            .clone()
-            .unwrap_or_default()
-            .split(',')
-            .map(|s| s.to_string())
-            .collect();
-
-        for url in webhook_urls {
-            status_update_plugins.push(Box::new(WebhookPlugin::new(url.to_string())));
-        }
 
         tasks.spawn(async move {
             let status_updater = NodeStatusUpdater::new(

crates/orchestrator/src/main.rs

@@ -1,4 +1,5 @@
 use prometheus::{GaugeVec, Opts, Registry, TextEncoder};
+pub mod webhook_sender;
 
 pub struct MetricsContext {
     pub compute_task_gauges: GaugeVec,