kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: improve webhook settings cache invalidation

Browse Source
This commit is contained in:
Ruben Fiszel
2025-02-24 16:10:53 +01:00
parent c5f3a59f2f
commit 5b79f4cf11
5 changed files with 40 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/migrations/20250205131521_clear_cache_on_webhook_changes.down.sql Normal file
View File

@@ -0,0 +1,4 @@
-- Add down migration script here
DROP TRIGGER webhook_change_trigger ON workspace_settings;
DROP FUNCTION notify_webhook_change();

15
backend/migrations/20250205131521_clear_cache_on_webhook_changes.up.sql Normal file
View File

@@ -0,0 +1,15 @@
-- Add up migration script here
CREATE OR REPLACE FUNCTION notify_webhook_change()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify('notify_webhook_change', NEW.workspace_id);
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER webhook_change_trigger
AFTER UPDATE OF webhook ON workspace_settings
FOR EACH ROW
WHEN (OLD.webhook IS DISTINCT FROM NEW.webhook)
EXECUTE FUNCTION notify_webhook_change();

11
backend/src/main.rs
View File

@@ -722,6 +722,11 @@ Windmill Community Edition {GIT_VERSION}
}
}
},
"notify_webhook_change" => {
let workspace_id = n.payload();
tracing::info!("Webhook change detected, invalidating webhook cache: {}", workspace_id);
windmill_api::webhook_util::WEBHOOK_CACHE.remove(workspace_id);
},
"notify_global_setting_change" => {
tracing::info!("Global setting change detected: {}", n.payload());
match n.payload() {
@@ -960,7 +965,11 @@ async fn listen_pg(db: &DB) -> Option<PgListener> {
};
if let Err(e) = listener
.listen_all(vec!["notify_config_change", "notify_global_setting_change"])
.listen_all(vec![
"notify_config_change",
"notify_global_setting_change",
"notify_webhook_change",
])
.await
{
tracing::error!(error = %e, "Could not listen to database");

2
backend/windmill-api/src/lib.rs
View File

@@ -116,7 +116,7 @@ mod users;
mod users_ee;
mod utils;
mod variables;
mod webhook_util;
pub mod webhook_util;
#[cfg(feature = "websocket")]
mod websocket_triggers;
mod workers;

15
backend/windmill-api/src/webhook_util.rs
View File

@@ -25,6 +25,8 @@ lazy_static::lazy_static! {
pub static ref INSTANCE_EVENTS_WEBHOOK: Option<String> = std::env::var("INSTANCE_EVENTS_WEBHOOK").ok();
pub static ref WEBHOOK_CACHE: Cache<String, Option<String>> = Cache::new(100);
}
pub enum WebhookPayload {
@@ -76,7 +78,6 @@ impl WebhookShared {
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let cache = Cache::new(100);
loop {
select! {
@@ -84,12 +85,12 @@ impl WebhookShared {
_ = shutdown_rx.recv() => break,
r = rx.recv() => match r {
Some(WebhookPayload::WorkspaceEvent(workspace_id, message)) => {
let webhook_opt = match cache.get(&workspace_id) {
let webhook_opt = match WEBHOOK_CACHE.get(&workspace_id) {
Some(guard) => {
guard
},
None => {
let Ok(webook_opt) =
let Ok(mut webhook_opt) =
sqlx::query_scalar!(
"SELECT webhook FROM workspace_settings WHERE workspace_id = $1",
workspace_id
@@ -101,13 +102,17 @@ impl WebhookShared {
tracing::error!("Webhook Message to send - but cannot get workspace settings! Workspace: {workspace_id}");
continue;
};
cache.insert(workspace_id, webook_opt.clone());
webook_opt
if webhook_opt.as_ref().is_some_and(|x| x.is_empty()) {
webhook_opt = None;
}
WEBHOOK_CACHE.insert(workspace_id, webhook_opt.clone());
webhook_opt
}
};
if let Some(url) = webhook_opt {
#[cfg(feature = "prometheus")]
let timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
tracing::info!("Sending webhook message to {}", url);
let _ = client.post(url).json(&message).send().await;
#[cfg(feature = "prometheus")]
timer.map(|x| x.stop_and_record());