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

add variable caching api

Browse Source
This commit is contained in:
Ruben Fiszel
2025-09-02 14:34:43 +00:00
parent f3fd1e90b0
commit 593da3213f
7 changed files with 73 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/migrations/20250902085503_notify_resource_cache_on_variable_change.up.sql
+3 -22
View File
@@ -1,23 +1,4 @@
-- Add up migration script here
CREATE OR REPLACE FUNCTION notify_var_resource_cache_change()
RETURNS TRIGGER AS $$
BEGIN
IF TG_TABLE_NAME = 'variable' THEN
PERFORM pg_notify('var_cache_invalidation',
json_build_object(
'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id),
'path', COALESCE(NEW.path, OLD.path),
'operation', TG_OP
)::text
);
END IF;
PERFORM pg_notify('resource_cache_invalidation',
json_build_object(
'workspace_id', COALESCE(NEW.workspace_id, OLD.workspace_id),
'path', COALESCE(NEW.path, OLD.path),
'operation', TG_OP
)::text
);
RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER variable_cache_invalidate_trigger ON variable;
DROP TRIGGER resource_cache_invalidate_trigger ON resource;
DROP FUNCTION notify_var_resource_cache_change();
backend/src/main.rs
-2
View File
@@ -1299,8 +1299,6 @@ async fn listen_pg(url: &str) -> Option<PgListener> {
"notify_workspace_key_change",
"notify_runnable_version_change",
"notify_token_invalidation",
"var_cache_invalidation",
"resource_cache_invalidation",
];
#[cfg(feature = "http_trigger")]
backend/windmill-api/openapi.yaml
+12 -1
View File
@@ -3254,6 +3254,12 @@ paths:
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
- name: allow_cache
description: |
allow getting a cached value for improved performance
in: query
schema:
type: boolean
responses:
"200":
description: variable
@@ -4017,6 +4023,11 @@ paths:
schema:
type: string
format: uuid
- name: allow_cache
description: allow getting a cached value for improved performance
in: query
schema:
type: boolean
responses:
"200":
description: resource value
@@ -15233,7 +15244,7 @@ components:
nu,
java,
ruby,
duckdb
duckdb,
# for related places search: ADD_NEW_LANG
]
backend/windmill-api/src/apps.rs
+1
View File
@@ -2517,6 +2517,7 @@ async fn build_args(
&path,
None,
"",
false
)
.await?;
if res.is_none() {
backend/windmill-api/src/resources.rs
+27 -35
View File
@@ -9,10 +9,7 @@
use std::collections::HashMap;
use crate::{
db::{ApiAuthed, DB},
users::{maybe_refresh_folders, require_owner_of_path, Tokened},
utils::check_scopes,
webhook_util::{WebhookMessage, WebhookShared},
db::{ApiAuthed, DB}, users::{maybe_refresh_folders, require_owner_of_path, Tokened}, utils::check_scopes, var_resource_cache::{cache_resource, get_cached_resource}, webhook_util::{WebhookMessage, WebhookShared}
};
use axum::{
body::Body,
@@ -36,7 +33,6 @@ use windmill_common::{
variables,
worker::CLOUD_HOSTED,
};
use crate::var_resource_cache::{get_cached_resource, cache_resource};
pub fn workspaced_service() -> Router {
Router::new()
@@ -327,28 +323,16 @@ async fn exists_resource(
Ok(Json(exists))
}
#[derive(Deserialize)]
struct GetResourceQuery {
allow_cache: Option<bool>,
}
async fn get_resource_value(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Query(q): Query<GetResourceQuery>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Option<serde_json::Value>> {
let path = path.to_path();
check_scopes(&authed, || format!("resources:read:{}", path))?;
// Check cache first when explicitly allowed
let allow_cache = q.allow_cache.unwrap_or(false);
if allow_cache {
if let Some(cached_value) = get_cached_resource(&w_id, &path) {
return Ok(Json(Some(cached_value)));
}
}
let mut tx = user_db.begin(&authed).await?;
@@ -367,12 +351,7 @@ async fn get_resource_value(
let value = not_found_if_none(value_o, "Resource", path)?;
// Cache the result if it exists and caching is allowed
if allow_cache {
if let Some(ref val) = value {
cache_resource(&w_id, &path, val.clone());
}
}
Ok(Json(value))
}
@@ -440,7 +419,9 @@ async fn custom_component(
#[derive(Deserialize)]
struct JobInfo {
job_id: Option<Uuid>,
allow_cache: Option<bool>,
}
async fn get_resource_value_interpolated(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
@@ -452,6 +433,7 @@ async fn get_resource_value_interpolated(
let path = path.to_path();
check_scopes(&authed, || format!("resources:read:{}", path))?;
return get_resource_value_interpolated_internal(
&authed,
Some(user_db),
@@ -460,6 +442,7 @@ async fn get_resource_value_interpolated(
path,
job_info.job_id,
token.as_str(),
job_info.allow_cache.unwrap_or(false),
)
.await
.map(|success| Json(success));
@@ -476,7 +459,13 @@ pub async fn get_resource_value_interpolated_internal(
path: &str,
job_id: Option<Uuid>,
token: &str,
allow_cache: bool,
) -> Result<Option<serde_json::Value>> {
if allow_cache {
if let Some(cached_value) = get_cached_resource(&workspace, &path) {
return Ok(Some(cached_value));
}
}
let mut tx = authed_transaction_or_default(authed, user_db.clone(), db).await?;
let value_o = sqlx::query_scalar!(
@@ -493,18 +482,20 @@ pub async fn get_resource_value_interpolated_internal(
let value = not_found_if_none(value_o, "Resource", path)?;
if let Some(value) = value {
Ok(Some(
transform_json_value(
authed,
user_db.clone(),
db,
workspace,
value,
&job_id,
token,
)
.await?,
))
let r = transform_json_value(
authed,
user_db.clone(),
db,
workspace,
value,
&job_id,
token,
)
.await?;
if allow_cache {
cache_resource(&workspace, &path, r.clone());
}
Ok(Some(r))
} else {
Ok(None)
}
@@ -540,6 +531,7 @@ pub async fn transform_json_value<'c>(
username_override: None,
token_prefix: None,
}),
false
)
.await?;
Ok(Value::String(v))
backend/windmill-api/src/var_resource_cache.rs
+7 -7
View File
@@ -9,10 +9,9 @@
use quick_cache::sync::Cache;
use serde_json::Value;
use std::time::{SystemTime, UNIX_EPOCH};
use windmill_common::variables::ListableVariable;
/// Cache TTL for variables and resources (60 seconds)
const CACHE_TTL_SECS: u64 = 60;
/// Cache TTL for variables and resources (30seconds)
const CACHE_TTL_SECS: u64 = 30;
/// Cache entry with timestamp and value (following raw script cache pattern)
#[derive(Clone, Debug)]
@@ -41,9 +40,10 @@ impl<T> CacheEntry<T> {
}
}
lazy_static::lazy_static! {
/// Cache for individual variable values: key = "workspace_id:path"
pub static ref VARIABLE_CACHE: Cache<String, CacheEntry<ListableVariable>> = Cache::new(1000);
pub static ref VARIABLE_CACHE: Cache<String, CacheEntry<String>> = Cache::new(1000);
/// Cache for resource values: key = "workspace_id:path"
pub static ref RESOURCE_CACHE: Cache<String, CacheEntry<Value>> = Cache::new(1000);
@@ -56,7 +56,7 @@ pub fn cache_key(workspace_id: &str, path: &str) -> String {
}
/// Get cached variable if available and not expired
pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option<ListableVariable> {
pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option<String> {
let key = cache_key(workspace_id, path);
VARIABLE_CACHE.get(&key).and_then(|entry| {
if entry.is_expired() {
@@ -70,8 +70,8 @@ pub fn get_cached_variable(workspace_id: &str, path: &str) -> Option<ListableVar
}
/// Cache variable data
pub fn cache_variable(workspace_id: &str, path: &str, variable: ListableVariable) {
let key = cache_key(workspace_id, path);
pub fn cache_variable(workspace_id: &str, path: &str, email: &str, variable: String) {
let key = format!("{}:{}", email, cache_key(workspace_id, path));
let entry = CacheEntry::new(variable);
VARIABLE_CACHE.insert(key.clone(), entry);
tracing::debug!("Cached variable {}", key);
backend/windmill-api/src/variables.rs
+23 -18
View File
@@ -135,7 +135,6 @@ async fn list_variables(
struct GetVariableQuery {
decrypt_secret: Option<bool>,
include_encrypted: Option<bool>,
allow_cache: Option<bool>,
}
async fn get_variable(
@@ -147,17 +146,6 @@ async fn get_variable(
) -> JsonResult<ListableVariable> {
let path = path.to_path();
check_scopes(&authed, || format!("variables:read:{}", path))?;
// Check cache first when explicitly allowed (and for appropriate requests)
let decrypt_secret = q.decrypt_secret.unwrap_or(true);
let allow_cache = q.allow_cache.unwrap_or(false);
let include_encrypted = q.include_encrypted.unwrap_or(false);
if allow_cache && (!decrypt_secret || include_encrypted) {
if let Some(cached_variable) = get_cached_variable(&w_id, &path) {
return Ok(Json(cached_variable));
}
}
let mut tx = user_db.begin(&authed).await?;
@@ -232,25 +220,25 @@ async fn get_variable(
variable
};
// Cache the result when explicitly allowed and caching appropriate
if allow_cache && (!decrypt_secret || include_encrypted) {
cache_variable(&w_id, &path, r.clone());
}
Ok(Json(r))
}
#[derive(Deserialize)]
struct GetValueQuery {
allow_cache: Option<bool>,
}
async fn get_value(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
Query(q): Query<GetValueQuery>,
) -> JsonResult<String> {
let path = path.to_path();
check_scopes(&authed, || format!("variables:read:{}", path))?;
let tx = user_db.begin(&authed).await?;
return get_value_internal(tx, &db, &w_id, &path, &authed)
return get_value_internal(tx, &db, &w_id, &path, &authed, q.allow_cache.unwrap_or(false))
.await
.map(Json);
}
@@ -712,7 +700,16 @@ pub async fn get_value_internal<'c>(
w_id: &str,
path: &str,
audit_author: &impl AuditAuthorable,
allow_cache: bool,
) -> Result<String> {
if allow_cache {
if let Some(cached_variable) = get_cached_variable(&w_id, &path) {
return Ok(cached_variable);
}
}
let variable_o = sqlx::query!(
"SELECT value, account, (now() > account.expires_at) as is_expired, is_secret, path from variable
LEFT JOIN account ON variable.account = account.id WHERE variable.path = $1 AND variable.workspace_id = $2", path, w_id
@@ -728,6 +725,8 @@ pub async fn get_value_internal<'c>(
unreachable!()
};
let r = if variable.is_secret {
audit_log(
&mut *tx,
@@ -765,6 +764,12 @@ pub async fn get_value_internal<'c>(
variable.value
};
// Cache the result when explicitly allowed and caching appropriate
if allow_cache {
cache_variable(&w_id, &path, audit_author.email(), r.clone());
}
Ok(r)
}