From 69d8d921232fa1eb0bb6d50893b70d01129ea84f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sun, 15 Feb 2026 15:32:15 +0100 Subject: [PATCH] make K8s operator private and add registry EE warning (#7955) Move K8s operator source code (crd, db_sync, reconciler, resolve) to windmill-ee-private and gate behind feature = "private". OSS stubs provide error messages when the feature is disabled. Add an info Alert banner in the Registries settings section when no enterprise license is active. Co-authored-by: Claude Opus 4.6 --- backend/Cargo.toml | 2 +- backend/ee-repo-ref.txt | 2 +- backend/windmill-operator/src/crd.rs | 218 ------------------ backend/windmill-operator/src/db_sync.rs | 52 ----- backend/windmill-operator/src/lib.rs | 22 +- backend/windmill-operator/src/operator_oss.rs | 19 ++ backend/windmill-operator/src/reconciler.rs | 179 -------------- backend/windmill-operator/src/resolve.rs | 126 ---------- .../lib/components/InstanceSettings.svelte | 7 + 9 files changed, 38 insertions(+), 589 deletions(-) delete mode 100644 backend/windmill-operator/src/crd.rs delete mode 100644 backend/windmill-operator/src/db_sync.rs create mode 100644 backend/windmill-operator/src/operator_oss.rs delete mode 100644 backend/windmill-operator/src/reconciler.rs delete mode 100644 backend/windmill-operator/src/resolve.rs diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 6987ce617f..bf3a18c486 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -96,7 +96,7 @@ lto = "thin" [features] default = [] -private = ["windmill-api/private", "windmill-api-agent-workers?/private", "windmill-autoscaling/private", "windmill-common/private", "windmill-git-sync/private", "windmill-indexer/private", "windmill-queue/private", "windmill-worker/private", "windmill-test-utils/private"] +private = ["windmill-api/private", "windmill-api-agent-workers?/private", "windmill-autoscaling/private", "windmill-common/private", "windmill-git-sync/private", "windmill-indexer/private", "windmill-operator?/private", "windmill-queue/private", "windmill-worker/private", "windmill-test-utils/private"] agent_worker_server = ["windmill-api/agent_worker_server", "dep:windmill-api-agent-workers", "windmill-test-utils/agent_worker_server"] enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-api-agent-workers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise"] local_reports = ["windmill-common/local_reports"] diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 0edac026f7..5cb666dcac 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -e7f80bca9320580e1cb96b4f4ca9942649abce7f \ No newline at end of file +c2f12c6989e9f069fd274085aefde29264f73c8c \ No newline at end of file diff --git a/backend/windmill-operator/src/crd.rs b/backend/windmill-operator/src/crd.rs deleted file mode 100644 index d832fea56f..0000000000 --- a/backend/windmill-operator/src/crd.rs +++ /dev/null @@ -1,218 +0,0 @@ -use std::collections::BTreeMap; - -use kube::CustomResource; -use schemars::JsonSchema; -use serde::{Deserialize, Serialize}; - -// Re-export all config types from windmill-common for downstream consumers. -pub use windmill_common::instance_config::{ - AutoscalingConfig, AutoscalingIntegration, CriticalErrorChannel, CustomInstanceDb, - CustomInstanceDbLogs, CustomInstancePgDatabases, DbOversizeAlert, Ducklake, DucklakeCatalog, - DucklakeCatalogResourceType, DucklakeSettings, DucklakeStorage, EnvRefWrapper, GlobalSettings, - IndexerSettings, OAuthClient, OAuthConfig, OtelSettings, OtelTracingProxySettings, ScriptLang, - SecretKeyRef, SecretKeyRefWrapper, SmtpSettings, StringOrSecretRef, TeamsChannel, - WorkerGroupConfig, -}; - -/// WindmillInstance CRD spec. -/// -/// Declares the desired state for instance-level configuration: -/// - `global_settings` maps directly to the `global_settings` table -/// - `worker_configs` maps to the `config` table with a `worker__` prefix -#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)] -#[kube( - group = "windmill.dev", - version = "v1alpha1", - kind = "WindmillInstance", - namespaced, - shortname = "wmi", - status = "WindmillInstanceStatus", - printcolumn = r#"{"name":"Synced","type":"string","jsonPath":".status.synced"}"#, - printcolumn = r#"{"name":"Last Synced","type":"date","jsonPath":".status.lastSyncedAt"}"#, - printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"# -)] -pub struct WindmillInstanceSpec { - /// Global settings to sync to the `global_settings` table. - #[serde(default)] - pub global_settings: GlobalSettings, - - /// Worker group configs to sync to the `config` table. - /// Keys are worker group names (e.g. "default", "gpu"). - /// Each key is stored in the DB as `worker__`. - #[serde(default)] - pub worker_configs: BTreeMap, -} - -// --------------------------------------------------------------------------- -// Status subresource -// --------------------------------------------------------------------------- - -/// Status subresource for WindmillInstance. -#[derive(Deserialize, Serialize, Clone, Debug, Default, JsonSchema)] -#[serde(rename_all = "camelCase")] -pub struct WindmillInstanceStatus { - /// Whether the last reconciliation was successful. - pub synced: bool, - /// Human-readable status message. - #[serde(default)] - pub message: String, - /// The `.metadata.generation` that was last observed. - #[serde(default)] - pub observed_generation: i64, - /// Timestamp of the last successful sync. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub last_synced_at: Option, -} - -#[cfg(test)] -mod tests { - use super::*; - use kube::CustomResourceExt; - - #[test] - fn crd_generation_produces_valid_yaml() { - let crd = WindmillInstance::crd(); - let yaml = serde_yml::to_string(&crd).expect("CRD should serialize to YAML"); - assert!( - yaml.contains("windmill.dev"), - "CRD should have group windmill.dev" - ); - assert!( - yaml.contains("v1alpha1"), - "CRD should have version v1alpha1" - ); - assert!( - yaml.contains("WindmillInstance"), - "CRD should have kind WindmillInstance" - ); - assert!(yaml.contains("wmi"), "CRD should have shortname wmi"); - } - - #[test] - fn crd_metadata() { - let crd = WindmillInstance::crd(); - assert_eq!( - crd.metadata.name.as_deref(), - Some("windmillinstances.windmill.dev") - ); - } - - #[test] - fn spec_deserializes_with_defaults() { - let json = r#"{"global_settings": {}, "worker_configs": {}}"#; - let spec: WindmillInstanceSpec = - serde_json::from_str(json).expect("Should deserialize empty spec"); - assert!(spec.global_settings.to_settings_map().is_empty()); - assert!(spec.worker_configs.is_empty()); - } - - #[test] - fn spec_deserializes_omitted_fields() { - let json = r#"{}"#; - let spec: WindmillInstanceSpec = - serde_json::from_str(json).expect("Should deserialize spec with missing fields"); - assert!(spec.global_settings.to_settings_map().is_empty()); - assert!(spec.worker_configs.is_empty()); - } - - #[test] - fn crd_schema_has_typed_properties() { - let crd = WindmillInstance::crd(); - let yaml = serde_yml::to_string(&crd).expect("CRD should serialize to YAML"); - assert!( - yaml.contains("base_url"), - "Schema should contain base_url property" - ); - assert!( - yaml.contains("smtp_settings"), - "Schema should contain smtp_settings property" - ); - assert!( - yaml.contains("worker_tags"), - "Schema should contain worker_tags property" - ); - assert!( - yaml.contains("retention_period_secs"), - "Schema should contain retention_period_secs property" - ); - assert!( - yaml.contains("otel_exporter_otlp_endpoint"), - "Schema should contain OTel endpoint property" - ); - assert!( - yaml.contains("min_workers"), - "Schema should contain autoscaling min_workers field" - ); - } - - #[test] - fn status_serializes_with_camel_case() { - let status = WindmillInstanceStatus { - synced: true, - message: "OK".to_string(), - observed_generation: 3, - last_synced_at: Some("2025-01-01T00:00:00Z".to_string()), - }; - let json = serde_json::to_value(&status).expect("Should serialize status"); - assert!(json.get("lastSyncedAt").is_some(), "Should use camelCase"); - assert!( - json.get("observedGeneration").is_some(), - "Should use camelCase" - ); - } - - #[test] - fn status_omits_null_last_synced_at() { - let status = WindmillInstanceStatus { - synced: false, - message: "Error".to_string(), - observed_generation: 1, - last_synced_at: None, - }; - let json = serde_json::to_value(&status).expect("Should serialize status"); - assert!( - json.get("lastSyncedAt").is_none(), - "Should omit null lastSyncedAt" - ); - } - - #[test] - fn status_default() { - let status = WindmillInstanceStatus::default(); - assert!(!status.synced); - assert!(status.message.is_empty()); - assert_eq!(status.observed_generation, 0); - assert!(status.last_synced_at.is_none()); - } - - #[test] - fn crd_schema_supports_secret_refs() { - let crd = WindmillInstance::crd(); - let yaml = serde_yml::to_string(&crd).expect("CRD should serialize to YAML"); - assert!( - yaml.contains("secretKeyRef"), - "CRD schema should contain secretKeyRef for secret reference support" - ); - } - - #[test] - fn spec_deserializes_secret_ref_fields() { - let json = r#"{ - "global_settings": { - "license_key": {"secretKeyRef": {"name": "wm-secrets", "key": "license"}}, - "base_url": "https://example.com" - } - }"#; - let spec: WindmillInstanceSpec = serde_json::from_str(json).unwrap(); - assert!(spec - .global_settings - .license_key - .as_ref() - .unwrap() - .is_secret_ref()); - assert_eq!( - spec.global_settings.base_url.as_deref(), - Some("https://example.com") - ); - } -} diff --git a/backend/windmill-operator/src/db_sync.rs b/backend/windmill-operator/src/db_sync.rs deleted file mode 100644 index 078cbbc9e7..0000000000 --- a/backend/windmill-operator/src/db_sync.rs +++ /dev/null @@ -1,52 +0,0 @@ -use std::collections::BTreeMap; - -use sqlx::{Pool, Postgres}; -use windmill_common::instance_config::{ - apply_configs_diff, apply_settings_diff, diff_global_settings, diff_worker_configs, ApplyMode, -}; - -/// Perform a full declarative sync of global settings. -/// -/// - Upserts every key in `desired` into the `global_settings` table. -/// - Deletes keys that exist in DB but are absent from `desired` -/// (except protected keys). -pub async fn sync_global_settings( - db: &Pool, - desired: &BTreeMap, -) -> anyhow::Result<()> { - // Fetch current settings from DB - let current_rows: Vec<(String, serde_json::Value)> = - sqlx::query_as("SELECT name, value FROM global_settings") - .fetch_all(db) - .await?; - let current: BTreeMap = current_rows.into_iter().collect(); - - let diff = diff_global_settings(¤t, desired, ApplyMode::Replace); - apply_settings_diff(db, &diff).await -} - -/// Perform a full declarative sync of worker configs. -/// -/// - Upserts every key in `desired` into the `config` table with the -/// `worker__` prefix. -/// - Deletes `worker__*` rows that exist in DB but are absent from `desired`. -pub async fn sync_worker_configs( - db: &Pool, - desired: &BTreeMap, -) -> anyhow::Result<()> { - // Fetch current worker configs from DB (strip prefix for comparison) - let current_rows: Vec<(String, serde_json::Value)> = - sqlx::query_as("SELECT name, config FROM config WHERE name LIKE 'worker__%'") - .fetch_all(db) - .await?; - let current: BTreeMap = current_rows - .into_iter() - .map(|(name, config)| { - let group = name.strip_prefix("worker__").unwrap_or(&name).to_string(); - (group, config) - }) - .collect(); - - let diff = diff_worker_configs(¤t, desired, ApplyMode::Replace); - apply_configs_diff(db, &diff).await -} diff --git a/backend/windmill-operator/src/lib.rs b/backend/windmill-operator/src/lib.rs index a82df4fe68..c82211d854 100644 --- a/backend/windmill-operator/src/lib.rs +++ b/backend/windmill-operator/src/lib.rs @@ -1,13 +1,11 @@ -pub mod crd; -pub mod db_sync; -pub mod reconciler; -pub mod resolve; +#[cfg(feature = "private")] +pub mod crd_ee; +#[cfg(feature = "private")] +pub mod db_sync_ee; +#[cfg(feature = "private")] +pub mod reconciler_ee; +#[cfg(feature = "private")] +pub mod resolve_ee; -pub use reconciler::run; - -/// Print the CRD YAML definition to stdout. -pub fn print_crd_yaml() { - use kube::CustomResourceExt; - let crd = crd::WindmillInstance::crd(); - println!("{}", serde_yml::to_string(&crd).unwrap()); -} +mod operator_oss; +pub use operator_oss::*; diff --git a/backend/windmill-operator/src/operator_oss.rs b/backend/windmill-operator/src/operator_oss.rs new file mode 100644 index 0000000000..d20cce548a --- /dev/null +++ b/backend/windmill-operator/src/operator_oss.rs @@ -0,0 +1,19 @@ +#[cfg(feature = "private")] +pub use crate::reconciler_ee::run; + +#[cfg(feature = "private")] +pub fn print_crd_yaml() { + use kube::CustomResourceExt; + let crd = crate::crd_ee::WindmillInstance::crd(); + println!("{}", serde_yml::to_string(&crd).unwrap()); +} + +#[cfg(not(feature = "private"))] +pub async fn run(_db: sqlx::Pool) -> anyhow::Result<()> { + anyhow::bail!("K8s operator is not available in this build") +} + +#[cfg(not(feature = "private"))] +pub fn print_crd_yaml() { + eprintln!("K8s operator CRD generation is not available in this build"); +} diff --git a/backend/windmill-operator/src/reconciler.rs b/backend/windmill-operator/src/reconciler.rs deleted file mode 100644 index fea355bed2..0000000000 --- a/backend/windmill-operator/src/reconciler.rs +++ /dev/null @@ -1,179 +0,0 @@ -use std::collections::BTreeMap; -use std::sync::Arc; -use std::time::Duration; - -use futures::StreamExt; -use kube::api::{Api, Patch, PatchParams}; -use kube::runtime::controller::Action; -use kube::runtime::Controller; -use kube::{Client, ResourceExt}; -use sqlx::{Pool, Postgres}; - -use crate::crd::{WindmillInstance, WindmillInstanceStatus}; -use crate::db_sync; -use crate::resolve; - -/// Shared state available to the reconciler. -struct Context { - db: Pool, - client: Client, -} - -/// Run the operator controller loop. Blocks until shutdown. -pub async fn run(db: Pool) -> anyhow::Result<()> { - let client = Client::try_default().await?; - - // Verify the CRD is installed by attempting to list - let api: Api = Api::all(client.clone()); - api.list(&Default::default()).await.map_err(|e| { - anyhow::anyhow!("Failed to list WindmillInstance CRDs. Is the CRD installed? Error: {e}") - })?; - tracing::info!("WindmillInstance CRD verified, starting controller"); - - let ctx = Arc::new(Context { db, client: client.clone() }); - - Controller::new(api, Default::default()) - .shutdown_on_signal() - .run(reconcile, error_policy, ctx) - .for_each(|res| async move { - match res { - Ok(o) => tracing::debug!("Reconciled: {:?}", o), - Err(e) => tracing::error!("Reconcile error: {:?}", e), - } - }) - .await; - - tracing::info!("Operator controller shut down"); - Ok(()) -} - -/// Main reconciliation logic for a single WindmillInstance resource. -async fn reconcile( - instance: Arc, - ctx: Arc, -) -> Result { - let name = instance.name_any(); - let ns = instance.namespace().unwrap_or_default(); - tracing::info!("Reconciling WindmillInstance {name} in namespace {ns}"); - - let generation = instance.metadata.generation.unwrap_or(0); - - // Resolve any secretKeyRef fields by reading K8s Secrets - let resolved = match resolve::resolve_secret_refs( - &ctx.client, - &ns, - &instance.spec.global_settings, - ) - .await - { - Ok(gs) => gs, - Err(e) => { - tracing::error!("Failed to resolve secret refs for {name}: {e:#}"); - update_status( - &ctx.client, - &instance, - false, - format!("Error resolving secret references: {e}"), - generation, - ) - .await?; - return Ok(Action::requeue(Duration::from_secs(30))); - } - }; - - // Convert typed structs to BTreeMaps for db_sync - let settings_map = resolved.to_settings_map(); - let configs_map: BTreeMap = instance - .spec - .worker_configs - .iter() - .map(|(k, v)| { - ( - k.clone(), - serde_json::to_value(v).expect("WorkerGroupConfig serialization cannot fail"), - ) - }) - .collect(); - - // Sync global settings - if let Err(e) = db_sync::sync_global_settings(&ctx.db, &settings_map).await { - tracing::error!("Failed to sync global settings for {name}: {e:#}"); - update_status( - &ctx.client, - &instance, - false, - format!("Error syncing global settings: {e}"), - generation, - ) - .await?; - // Requeue after 30s on error - return Ok(Action::requeue(Duration::from_secs(30))); - } - - // Sync worker configs - if let Err(e) = db_sync::sync_worker_configs(&ctx.db, &configs_map).await { - tracing::error!("Failed to sync worker configs for {name}: {e:#}"); - update_status( - &ctx.client, - &instance, - false, - format!("Error syncing worker configs: {e}"), - generation, - ) - .await?; - return Ok(Action::requeue(Duration::from_secs(30))); - } - - tracing::info!("Successfully synced WindmillInstance {name}"); - update_status( - &ctx.client, - &instance, - true, - "Synced successfully".to_string(), - generation, - ) - .await?; - - // Periodic re-sync every 5 minutes for drift detection - Ok(Action::requeue(Duration::from_secs(300))) -} - -/// Error policy: requeue after 60 seconds on unhandled errors. -fn error_policy( - _instance: Arc, - _error: &kube::Error, - _ctx: Arc, -) -> Action { - Action::requeue(Duration::from_secs(60)) -} - -/// Patch the status subresource of the WindmillInstance. -async fn update_status( - client: &Client, - instance: &WindmillInstance, - synced: bool, - message: String, - observed_generation: i64, -) -> Result<(), kube::Error> { - let name = instance.name_any(); - let ns = instance.namespace().unwrap_or_default(); - let api: Api = Api::namespaced(client.clone(), &ns); - - let now = chrono::Utc::now().to_rfc3339(); - let status = WindmillInstanceStatus { - synced, - message, - observed_generation, - last_synced_at: if synced { Some(now) } else { None }, - }; - - let patch = serde_json::json!({ "status": status }); - api.patch_status( - &name, - &PatchParams::apply("windmill-operator"), - &Patch::Merge(&patch), - ) - .await?; - - Ok(()) -} diff --git a/backend/windmill-operator/src/resolve.rs b/backend/windmill-operator/src/resolve.rs deleted file mode 100644 index b6867b984d..0000000000 --- a/backend/windmill-operator/src/resolve.rs +++ /dev/null @@ -1,126 +0,0 @@ -use std::collections::BTreeMap; - -use k8s_openapi::api::core::v1::Secret; -use kube::api::Api; -use kube::Client; -use windmill_common::instance_config::{GlobalSettings, SecretKeyRef, StringOrSecretRef}; - -#[derive(Debug, thiserror::Error)] -pub enum ResolveError { - #[error("failed to fetch Kubernetes Secret '{name}': {source}")] - FetchSecret { name: String, source: kube::Error }, - #[error("key '{key}' not found in Secret '{secret}'")] - KeyNotFound { secret: String, key: String }, - #[error("value for key '{key}' in Secret '{secret}' is not valid UTF-8")] - InvalidUtf8 { secret: String, key: String }, - #[error("environment variable '{var}' not found")] - EnvVarNotFound { var: String }, -} - -/// Resolve all `StringOrSecretRef` fields in `GlobalSettings` by reading -/// referenced Kubernetes Secrets. Returns a new `GlobalSettings` with every -/// `SecretRef` replaced by its `Literal` value. -pub async fn resolve_secret_refs( - client: &Client, - namespace: &str, - settings: &GlobalSettings, -) -> Result { - let mut settings = settings.clone(); - let mut cache: BTreeMap> = BTreeMap::new(); - - resolve_option(client, namespace, &mut cache, &mut settings.license_key).await?; - resolve_option(client, namespace, &mut cache, &mut settings.hub_api_secret).await?; - resolve_option(client, namespace, &mut cache, &mut settings.scim_token).await?; - - if let Some(smtp) = &mut settings.smtp_settings { - resolve_option(client, namespace, &mut cache, &mut smtp.smtp_password).await?; - } - - if let Some(oauths) = &mut settings.oauths { - for oauth in oauths.values_mut() { - resolve_field(client, namespace, &mut cache, &mut oauth.secret).await?; - } - } - - if let Some(pg) = &mut settings.custom_instance_pg_databases { - resolve_option(client, namespace, &mut cache, &mut pg.user_pwd).await?; - } - - Ok(settings) -} - -async fn resolve_option( - client: &Client, - namespace: &str, - cache: &mut BTreeMap>, - field: &mut Option, -) -> Result<(), ResolveError> { - if let Some(val) = field { - resolve_field(client, namespace, cache, val).await?; - } - Ok(()) -} - -async fn resolve_field( - client: &Client, - namespace: &str, - cache: &mut BTreeMap>, - field: &mut StringOrSecretRef, -) -> Result<(), ResolveError> { - if let Some(var_name) = field.as_env_ref() { - let var_name = var_name.to_string(); - let value = - std::env::var(&var_name).map_err(|_| ResolveError::EnvVarNotFound { var: var_name })?; - *field = StringOrSecretRef::Literal(value); - return Ok(()); - } - - let secret_ref = match field.as_secret_ref() { - Some(r) => r.clone(), - None => return Ok(()), - }; - - let value = fetch_secret_value(client, namespace, cache, &secret_ref).await?; - *field = StringOrSecretRef::Literal(value); - Ok(()) -} - -async fn fetch_secret_value( - client: &Client, - namespace: &str, - cache: &mut BTreeMap>, - secret_ref: &SecretKeyRef, -) -> Result { - if let Some(data) = cache.get(&secret_ref.name) { - return data - .get(&secret_ref.key) - .cloned() - .ok_or_else(|| ResolveError::KeyNotFound { - secret: secret_ref.name.clone(), - key: secret_ref.key.clone(), - }); - } - - let api: Api = Api::namespaced(client.clone(), namespace); - let secret = api - .get(&secret_ref.name) - .await - .map_err(|e| ResolveError::FetchSecret { name: secret_ref.name.clone(), source: e })?; - - let data = secret.data.unwrap_or_default(); - let decoded: BTreeMap = data - .into_iter() - .filter_map(|(k, v)| String::from_utf8(v.0).ok().map(|s| (k, s))) - .collect(); - - let value = decoded - .get(&secret_ref.key) - .cloned() - .ok_or_else(|| ResolveError::KeyNotFound { - secret: secret_ref.name.clone(), - key: secret_ref.key.clone(), - })?; - - cache.insert(secret_ref.name.clone(), decoded); - Ok(value) -} diff --git a/frontend/src/lib/components/InstanceSettings.svelte b/frontend/src/lib/components/InstanceSettings.svelte index 8b352d1e2b..9bf318c4ce 100644 --- a/frontend/src/lib/components/InstanceSettings.svelte +++ b/frontend/src/lib/components/InstanceSettings.svelte @@ -851,6 +851,13 @@ description="Add private registries for Pip, Bun and npm." link="https://www.windmill.dev/docs/advanced/imports" /> + {#if !$enterpriseLicense} + + {/if} {:else if category == 'Alerts'}