From 85e460d853cbd9f8d245efb9009690c0bb468bfc Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 23 Jan 2026 12:23:07 +0000 Subject: [PATCH] feat: add cache-rt command and SYNC_CACHED_RT env variable for resource types (#7666) * feat: add cache-rt command and SYNC_CACHED_RT env variable for resource types Add a new CLI command `cache-rt` that fetches resource types from the hub and caches them locally. This command is run during Docker image build to pre-populate the cache. Add `SYNC_CACHED_RT` environment variable that, when set to true, syncs cached resource types to the admins workspace's resource_type table on server startup. The sync is incremental - it only updates resource types that have changed. Changes: - Add HUB_RT_CACHE_DIR constant for resource type cache location - Add cache-rt CLI command to fetch and cache resource types from hub - Add sync_cached_resource_types() to sync cached types to database - Update all Dockerfiles to run cache-rt during image build Co-Authored-By: Claude Opus 4.5 * fix: parse schema string from hub into proper JSONB The hub API returns schema as a JSON string that needs to be parsed into a proper JSON value before inserting into the database. Co-Authored-By: Claude Opus 4.5 --------- Co-authored-by: Claude Opus 4.5 --- Dockerfile | 2 +- backend/src/main.rs | 176 +++++++++++++++++++++++++- backend/windmill-common/src/worker.rs | 1 + docker/DockerfileSlim | 2 + docker/DockerfileSlimEe | 2 + 5 files changed, 180 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index 4000c3a537..e1e30876c4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -254,7 +254,7 @@ COPY ./frontend/src/lib/hubPaths.json ${APP}/hubPaths.json RUN windmill cache ${APP}/hubPaths.json && rm ${APP}/hubPaths.json - +RUN windmill cache-rt # Create a non-root user 'windmill' with UID and GID 1000 RUN addgroup --gid 1000 windmill && \ diff --git a/backend/src/main.rs b/backend/src/main.rs index e789cfb767..e85565dfd8 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -59,9 +59,10 @@ use windmill_common::{ MODE_AND_ADDONS, }, worker::{ - reload_custom_tags_setting, Connection, HUB_CACHE_DIR, TMP_DIR, TMP_LOGS_DIR, WORKER_GROUP, + reload_custom_tags_setting, Connection, HUB_CACHE_DIR, HUB_RT_CACHE_DIR, TMP_DIR, + TMP_LOGS_DIR, WORKER_GROUP, }, - KillpillSender, METRICS_ENABLED, + KillpillSender, DEFAULT_HUB_BASE_URL, METRICS_ENABLED, }; #[cfg(feature = "enterprise")] @@ -319,6 +320,160 @@ async fn cache_hub_scripts(file_path: Option) -> anyhow::Result<()> { Ok(()) } +/// Raw resource type from hub API (schema is a JSON string) +#[derive(serde::Deserialize)] +struct HubResourceTypeRaw { + pub id: i64, + pub name: String, + pub schema: Option, + pub app: String, + pub description: Option, +} + +/// Processed resource type with parsed schema +#[derive(serde::Deserialize, serde::Serialize, Clone)] +pub struct HubResourceType { + pub id: i64, + pub name: String, + pub schema: Option, + pub app: String, + pub description: Option, +} + +const HUB_RT_CACHE_FILE: &str = "resource_types.json"; + +async fn cache_hub_resource_types() -> anyhow::Result<()> { + println!("Caching resource types from hub..."); + + let response = HTTP_CLIENT + .get(format!("{}/resource_types/list", DEFAULT_HUB_BASE_URL)) + .header("Accept", "application/json") + .send() + .await + .with_context(|| "Failed to fetch resource types from hub")?; + + if !response.status().is_success() { + anyhow::bail!( + "Failed to fetch resource types from hub: {}", + response.status() + ); + } + + let raw_types: Vec = response + .json::>() + .await + .with_context(|| "Failed to parse resource types from hub")?; + + // Parse schema strings into JSON values + let resource_types: Vec = raw_types + .into_iter() + .filter_map(|rt| { + let schema = match rt.schema { + Some(s) => match serde_json::from_str(&s) { + Ok(v) => Some(v), + Err(e) => { + println!("Warning: failed to parse schema for {}: {}", rt.name, e); + return None; + } + }, + None => None, + }; + Some(HubResourceType { + id: rt.id, + name: rt.name, + schema, + app: rt.app, + description: rt.description, + }) + }) + .collect(); + + println!("Fetched {} resource types from hub", resource_types.len()); + + create_dir_all(HUB_RT_CACHE_DIR)?; + + let cache_path = format!("{}/{}", HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE); + let content = serde_json::to_string_pretty(&resource_types) + .with_context(|| "Failed to serialize resource types")?; + + std::fs::write(&cache_path, content) + .with_context(|| format!("Failed to write cache file to {}", cache_path))?; + + println!("Cached resource types to {}", cache_path); + Ok(()) +} + +pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyhow::Result<()> { + let cache_path = format!("{}/{}", HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE); + + if tokio::fs::metadata(&cache_path).await.is_err() { + tracing::info!("No cached resource types found at {}, skipping sync", cache_path); + return Ok(()); + } + + tracing::info!("Syncing cached resource types to admins workspace..."); + + let content = tokio::fs::read_to_string(&cache_path) + .await + .with_context(|| format!("Failed to read cache file from {}", cache_path))?; + + let cached_types: Vec = serde_json::from_str(&content) + .with_context(|| "Failed to parse cached resource types")?; + + tracing::info!("Found {} cached resource types", cached_types.len()); + + // Get existing resource types in admins workspace + let existing_types: Vec<(String, Option, Option)> = sqlx::query_as( + "SELECT name, schema, description FROM resource_type WHERE workspace_id = 'admins'", + ) + .fetch_all(db) + .await + .with_context(|| "Failed to fetch existing resource types")?; + + let existing_map: std::collections::HashMap, Option)> = + existing_types + .into_iter() + .map(|(name, schema, desc)| (name, (schema, desc))) + .collect(); + + let mut synced_count = 0; + let mut skipped_count = 0; + + for rt in cached_types { + // Check if resource type already exists with same schema and description + if let Some((existing_schema, existing_desc)) = existing_map.get(&rt.name) { + if existing_schema == &rt.schema && existing_desc == &rt.description { + skipped_count += 1; + continue; + } + } + + // Insert or update resource type + sqlx::query( + "INSERT INTO resource_type (workspace_id, name, schema, description, edited_at) + VALUES ('admins', $1, $2, $3, now()) + ON CONFLICT (workspace_id, name) DO UPDATE + SET schema = EXCLUDED.schema, description = EXCLUDED.description, edited_at = now()", + ) + .bind(&rt.name) + .bind(&rt.schema) + .bind(&rt.description) + .execute(db) + .await + .with_context(|| format!("Failed to upsert resource type {}", rt.name))?; + + synced_count += 1; + } + + tracing::info!( + "Synced {} resource types to admins workspace ({} skipped as unchanged)", + synced_count, + skipped_count + ); + + Ok(()) +} + fn print_help() { println!("Windmill - a fast, open-source workflow engine and job runner."); println!(); @@ -329,6 +484,7 @@ fn print_help() { println!(" help | -h | --help Show this help information and exit"); println!(" version Show Windmill version and exit"); println!(" cache [hubPaths.json] Pre-cache hub scripts (default: ./hubPaths.json)"); + println!(" cache-rt Pre-cache hub resource types"); println!(); println!("Environment variables (name = default):"); println!(" DATABASE_URL = The Postgres database url."); @@ -344,6 +500,7 @@ fn print_help() { println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)"); println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup"); println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool"); + println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start"); println!(); println!("Notes:"); println!("- Advanced and less commonly used settings are managed via the database and are omitted here."); @@ -438,6 +595,10 @@ async fn windmill_main() -> anyhow::Result<()> { windmill_worker::run_prepare_deps_cli().await?; return Ok(()); } + "cache-rt" => { + cache_hub_resource_types().await?; + return Ok(()); + } _ => {} } @@ -542,6 +703,17 @@ async fn windmill_main() -> anyhow::Result<()> { } else { tracing::info!("SKIP_MIGRATION set, skipping db migration...") } + + // Sync cached resource types to admins workspace if SYNC_CACHED_RT is set + if std::env::var("SYNC_CACHED_RT") + .ok() + .map(|v| v.to_lowercase() == "true" || v == "1") + .unwrap_or(false) + { + if let Err(e) = sync_cached_resource_types(db).await { + tracing::warn!("Failed to sync cached resource types: {:#}", e); + } + } } } diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index bc0e20cd35..9949fc8792 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -486,6 +486,7 @@ pub const TMP_DIR: &str = "/tmp/windmill"; pub const TMP_LOGS_DIR: &str = concatcp!(TMP_DIR, "/logs"); pub const HUB_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub"); +pub const HUB_RT_CACHE_DIR: &str = concatcp!(ROOT_CACHE_DIR, "hub_rt"); pub const ROOT_CACHE_DIR: &str = concatcp!(TMP_DIR, "/cache/"); diff --git a/docker/DockerfileSlim b/docker/DockerfileSlim index a4d1569019..bd07238b4c 100644 --- a/docker/DockerfileSlim +++ b/docker/DockerfileSlim @@ -51,6 +51,8 @@ RUN windmill cache ${APP}/hubPaths.json RUN rm ${APP}/hubPaths.json +RUN windmill cache-rt + # Create directories and make world-accessible for arbitrary UID support RUN mkdir -p -m 777 /tmp/windmill/logs /tmp/windmill/search /tmp/.cache && \ chmod 777 /tmp/.cache && \ diff --git a/docker/DockerfileSlimEe b/docker/DockerfileSlimEe index 4417566c4f..b127d1b782 100644 --- a/docker/DockerfileSlimEe +++ b/docker/DockerfileSlimEe @@ -51,6 +51,8 @@ RUN windmill cache ${APP}/hubPaths.json RUN rm ${APP}/hubPaths.json +RUN windmill cache-rt + # Create directories and make world-accessible for arbitrary UID support RUN mkdir -p -m 777 /tmp/windmill/logs /tmp/windmill/search /tmp/.cache && \ chmod 777 /tmp/.cache && \