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

make K8s operator private and add registry EE warning (#7955)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-02-15 15:32:15 +01:00
committed by GitHub
parent 1754a64cd6
commit 69d8d92123
9 changed files with 38 additions and 589 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/Cargo.toml
View File

@@ -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"]

2
backend/ee-repo-ref.txt
View File

@@ -1 +1 @@
e7f80bca9320580e1cb96b4f4ca9942649abce7f
c2f12c6989e9f069fd274085aefde29264f73c8c

218
backend/windmill-operator/src/crd.rs
View File

@@ -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__<key>`.
#[serde(default)]
pub worker_configs: BTreeMap<String, WorkerGroupConfig>,
}
// ---------------------------------------------------------------------------
// 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<String>,
}
#[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")
);
}
}

52
backend/windmill-operator/src/db_sync.rs
View File

@@ -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<Postgres>,
desired: &BTreeMap<String, serde_json::Value>,
) -> 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<String, serde_json::Value> = current_rows.into_iter().collect();
let diff = diff_global_settings(&current, 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<Postgres>,
desired: &BTreeMap<String, serde_json::Value>,
) -> 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<String, serde_json::Value> = 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(&current, desired, ApplyMode::Replace);
apply_configs_diff(db, &diff).await
}

22
backend/windmill-operator/src/lib.rs
View File

@@ -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::*;

19
backend/windmill-operator/src/operator_oss.rs Normal file
View File

@@ -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<sqlx::Postgres>) -> 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");
}

179
backend/windmill-operator/src/reconciler.rs
View File

@@ -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<Postgres>,
client: Client,
}
/// Run the operator controller loop. Blocks until shutdown.
pub async fn run(db: Pool<Postgres>) -> anyhow::Result<()> {
let client = Client::try_default().await?;
// Verify the CRD is installed by attempting to list
let api: Api<WindmillInstance> = 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<WindmillInstance>,
ctx: Arc<Context>,
) -> Result<Action, kube::Error> {
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<String, serde_json::Value> = 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<WindmillInstance>,
_error: &kube::Error,
_ctx: Arc<Context>,
) -> 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<WindmillInstance> = 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(())
}

126
backend/windmill-operator/src/resolve.rs
View File

@@ -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<GlobalSettings, ResolveError> {
let mut settings = settings.clone();
let mut cache: BTreeMap<String, BTreeMap<String, String>> = 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<String, BTreeMap<String, String>>,
field: &mut Option<StringOrSecretRef>,
) -> 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<String, BTreeMap<String, String>>,
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<String, BTreeMap<String, String>>,
secret_ref: &SecretKeyRef,
) -> Result<String, ResolveError> {
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<Secret> = 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<String, String> = 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)
}

7
frontend/src/lib/components/InstanceSettings.svelte
View File

@@ -851,6 +851,13 @@
description="Add private registries for Pip, Bun and npm."
link="https://www.windmill.dev/docs/advanced/imports"
/>
{#if !$enterpriseLicense}
<Alert
type="info"
title="Private registries configuration is an EE feature"
class="mb-2"
/>
{/if}
{:else if category == 'Alerts'}
<SettingsPageHeader
title="Alerts"