From c1c8ead150d18d98deab4ef6dc9c18076d395204 Mon Sep 17 00:00:00 2001 From: Andrew Kenworthy Date: Fri, 18 Sep 2026 17:34:34 +0200 Subject: [PATCH] refactor: Flatten role configs --- rust/operator-binary/src/controller.rs | 84 +++++++-- .../src/controller/build/mod.rs | 167 ++++++++++-------- .../controller/build/resource/config_map.rs | 5 +- .../controller/build/resource/statefulset.rs | 41 ++--- .../src/controller/validate.rs | 144 +++++++++------ rust/operator-binary/src/crd/mod.rs | 18 -- 6 files changed, 260 insertions(+), 199 deletions(-) diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 18f283c1..b7b842a9 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -13,6 +13,7 @@ use stackable_operator::{ cluster_resources::ClusterResourceApplyStrategy, commons::{ affinity::StackableAffinity, + pdb::PdbConfig, product_image_selection::ResolvedProductImage, resources::{NoRuntimeLimits, Resources}, }, @@ -107,12 +108,27 @@ pub struct KubernetesResources { pub status: PhantomData, } -/// Per-role configuration extracted during validation. +/// The Node role's validated role-level configuration. +/// +/// Separate from the other roles' because only the Node role serves the web UI, so only it has a +/// listener class and a group listener. One shared type would have to make both fields `Option` — +/// mandatory for the Node role, meaningless for Worker and Beat — leaving every reader to work out +/// which role it is holding. +#[derive(Clone, Debug)] +pub struct ValidatedNodeRoleConfig { + pub pdb: PdbConfig, + pub listener_class: ListenerClassName, + pub group_listener_name: ListenerName, +} + +/// The Worker and Beat roles' validated role-level configuration. +/// +/// Neither serves the web UI, so there is no listener class to carry and the Pod disruption budget +/// is all that is left. Structurally identical to `GenericRoleConfig`; kept as its own type so the +/// validated cluster holds controller-owned types throughout. #[derive(Clone, Debug)] pub struct ValidatedRoleConfig { - pub pdb: Option, - pub listener_class: Option, - pub group_listener_name: Option, + pub pdb: PdbConfig, } /// A validated, merged Superset role-group config. @@ -201,22 +217,52 @@ pub struct ValidatedCluster { pub product_version: ProductVersion, pub image: ResolvedProductImage, pub cluster_config: ValidatedClusterConfig, - pub role_groups: BTreeMap>, - pub role_configs: BTreeMap, + pub node_config: ValidatedNodeRoleConfig, + pub node_role_group_configs: BTreeMap, + pub worker_config: Option, + pub worker_role_group_configs: BTreeMap, + pub beat_config: Option, + pub beat_role_group_configs: BTreeMap, +} + +/// The non-derived inputs to [`ValidatedCluster::new`]. +/// +/// Named fields, so the three same-typed role-group maps — and the two +/// `Option` — cannot be swapped silently. +#[derive(Debug)] +pub struct ValidatedClusterParams { + pub name: ClusterName, + pub namespace: NamespaceName, + pub uid: Uid, + pub image: ResolvedProductImage, + pub cluster_config: ValidatedClusterConfig, + pub node_config: ValidatedNodeRoleConfig, + pub node_role_group_configs: BTreeMap, + pub worker_config: Option, + pub worker_role_group_configs: BTreeMap, + pub beat_config: Option, + pub beat_role_group_configs: BTreeMap, } impl ValidatedCluster { - pub fn new( - name: ClusterName, - namespace: NamespaceName, - uid: Uid, - image: ResolvedProductImage, - cluster_config: ValidatedClusterConfig, - role_groups: BTreeMap>, - role_configs: BTreeMap, - ) -> Self { + pub fn new(params: ValidatedClusterParams) -> Self { + let ValidatedClusterParams { + name, + namespace, + uid, + image, + cluster_config, + node_config, + node_role_group_configs, + worker_config, + worker_role_group_configs, + beat_config, + beat_role_group_configs, + } = params; + let product_version = ProductVersion::from_str(&image.app_version_label_value) .expect("the app version label value is a valid product version"); + Self { // Capture only the identity fields needed to own child objects, derived from the // typed cluster identity rather than the raw CRD. @@ -228,8 +274,12 @@ impl ValidatedCluster { }, image, cluster_config, - role_groups, - role_configs, + node_config, + node_role_group_configs, + worker_config, + worker_role_group_configs, + beat_config, + beat_role_group_configs, name, namespace, uid, diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index b0861e09..e485f7f4 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -5,6 +5,7 @@ use std::marker::PhantomData; use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::meta::ObjectMetaBuilder, + k8s_openapi::api::core::v1::{ConfigMap, Service}, kvp::Labels, v2::{ builder::meta::ownerreference_from_resource, @@ -16,7 +17,7 @@ use stackable_operator::{ use crate::{ controller::{ CONTROLLER_NAME, KubernetesResources, OPERATOR_NAME, PRODUCT_NAME, Prepared, - ValidatedCluster, + SupersetRoleGroupConfig, ValidatedCluster, build::resource::{ config_map::build_rolegroup_config_map, deployment::build_rolegroup_deployment, @@ -64,89 +65,74 @@ pub fn build(cluster: &ValidatedCluster) -> Result let mut config_maps = vec![]; let mut pod_disruption_budgets = vec![]; - for (superset_role, role_group_configs) in &cluster.role_groups { - for (role_group_name, rolegroup_config) in role_group_configs { - let config = &rolegroup_config.config; - - config_maps.push( - build_rolegroup_config_map( - cluster, - superset_role, - role_group_name, - config, - &rolegroup_config.config_overrides, - ) - .context(ConfigMapSnafu { + for (role_group_name, rolegroup_config) in &cluster.node_role_group_configs { + let (config_map, metrics_service) = build_common_role_group_resources( + cluster, + &SupersetRole::Node, + role_group_name, + rolegroup_config, + )?; + config_maps.push(config_map); + services.push(metrics_service); + + // Only the Node role's StatefulSet references a headless Service (as its `serviceName`); + // the Worker/Beat Deployments have no `serviceName` and do not serve the HTTP port. + services.push(build_rolegroup_headless_service( + cluster, + &SupersetRole::Node, + role_group_name, + )); + + stateful_sets.push( + build_node_rolegroup_statefulset(cluster, role_group_name, rolegroup_config).context( + StatefulSetSnafu { role_group: role_group_name.clone(), - })?, - ); + }, + )?, + ); + } - // Every role exposes metrics via the statsd-exporter sidecar, so each rolegroup gets a - // metrics Service. - services.push(build_rolegroup_metrics_service( + // The Celery roles differ from each other only in name: both produce a ConfigMap, a metrics + // Service and a Deployment. + for (role, role_group_configs) in [ + (&SupersetRole::Worker, &cluster.worker_role_group_configs), + (&SupersetRole::Beat, &cluster.beat_role_group_configs), + ] { + for (role_group_name, rolegroup_config) in role_group_configs { + let (config_map, metrics_service) = build_common_role_group_resources( cluster, - superset_role, + role, role_group_name, - )); - - match superset_role { - SupersetRole::Node => { - // Only the `Node` role's StatefulSet references a headless Service (as its - // `serviceName`); the `Worker`/`Beat` Deployments have no `serviceName` and do - // not serve the HTTP port, so they get no headless Service. - services.push(build_rolegroup_headless_service( - cluster, - superset_role, - role_group_name, - )); - - stateful_sets.push( - build_node_rolegroup_statefulset( - cluster, - superset_role, - role_group_name, - rolegroup_config, - ) - .context(StatefulSetSnafu { - role_group: role_group_name.clone(), - })?, - ); - } - SupersetRole::Worker | SupersetRole::Beat => { - deployments.push( - build_rolegroup_deployment( - cluster, - superset_role, - role_group_name, - rolegroup_config, - ) - .context(DeploymentSnafu { - role_group: role_group_name.clone(), - })?, - ); - } - } + rolegroup_config, + )?; + config_maps.push(config_map); + services.push(metrics_service); + + deployments.push( + build_rolegroup_deployment(cluster, role, role_group_name, rolegroup_config) + .context(DeploymentSnafu { + role_group: role_group_name.clone(), + })?, + ); } + } - // Role-level resources (group listener, PDB) are built once per role, after its role - // groups — not once per role group. - if let Some(role_config) = cluster.role_configs.get(superset_role) { - if let (Some(listener_class), Some(listener_group_name)) = ( - &role_config.listener_class, - &role_config.group_listener_name, - ) { - listeners.push(build_group_listener( - cluster, - superset_role, - listener_class, - listener_group_name.to_string(), - )); - } - - if let Some(pdb_config) = &role_config.pdb { - pod_disruption_budgets.extend(build_pdb(pdb_config, cluster, superset_role)); - } - } + listeners.push(build_group_listener( + cluster, + &SupersetRole::Node, + &cluster.node_config.listener_class, + cluster.node_config.group_listener_name.to_string(), + )); + pod_disruption_budgets.extend(build_pdb( + &cluster.node_config.pdb, + cluster, + &SupersetRole::Node, + )); + if let Some(worker) = &cluster.worker_config { + pod_disruption_budgets.extend(build_pdb(&worker.pdb, cluster, &SupersetRole::Worker)); + } + if let Some(beat) = &cluster.beat_config { + pod_disruption_budgets.extend(build_pdb(&beat.pdb, cluster, &SupersetRole::Beat)); } Ok(KubernetesResources { @@ -162,6 +148,31 @@ pub fn build(cluster: &ValidatedCluster) -> Result }) } +/// The resources every role group gets regardless of its role: a ConfigMap and a metrics Service +/// (every role exposes metrics via the statsd-exporter sidecar). +fn build_common_role_group_resources( + cluster: &ValidatedCluster, + role: &SupersetRole, + role_group_name: &RoleGroupName, + rolegroup_config: &SupersetRoleGroupConfig, +) -> Result<(ConfigMap, Service), Error> { + let config_map = build_rolegroup_config_map( + cluster, + role, + role_group_name, + &rolegroup_config.config, + &rolegroup_config.config_overrides, + ) + .context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?; + + Ok(( + config_map, + build_rolegroup_metrics_service(cluster, role, role_group_name), + )) +} + /// Returns an [`ObjectMetaBuilder`] pre-filled with the namespace, an owner reference back to /// the cluster, and the recommended labels for a resource named `name` in `role`/ /// `role_group_name`. diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index e2343f6e..d3952f8d 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -120,9 +120,8 @@ mod tests { let role_group_name: RoleGroupName = "default".parse().expect("valid role group name"); let rolegroup_config = validated - .role_groups - .get(&SupersetRole::Node) - .and_then(|groups| groups.get(&role_group_name)) + .node_role_group_configs + .get(&role_group_name) .expect("node default rolegroup"); let config_map = build_rolegroup_config_map( diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 1cc17d5e..1a7e54f7 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -88,10 +88,13 @@ type Result = std::result::Result; /// The rolegroup [`StatefulSet`] runs the rolegroup, as configured by the administrator. pub fn build_node_rolegroup_statefulset( validated: &ValidatedCluster, - superset_role: &SupersetRole, role_group_name: &RoleGroupName, rolegroup_config: &SupersetRoleGroupConfig, ) -> Result { + // This function builds the Node role's StatefulSet only; the Worker and Beat roles get + // Deployments instead. + let superset_role = &SupersetRole::Node; + let merged_config = &rolegroup_config.config; let resource_names = validated.role_group_resource_names(superset_role, role_group_name); @@ -210,23 +213,13 @@ pub fn build_node_rolegroup_statefulset( .readiness_probe(readiness_probe) .liveness_probe(liveness_probe); - // listener endpoints will use persistent volumes - // so that load balancers can hard-code the target addresses and - // that it is possible to connect to a consistent address - let pvcs = if let Some(group_listener_name) = validated - .role_configs - .get(superset_role) - .and_then(|role_config| role_config.group_listener_name.clone()) - { - let pvc = listener_operator_volume_source_builder_build_pvc( - &ListenerReference::Listener(group_listener_name), - &unversioned_recommended_labels, - &super::LISTENER_VOLUME_NAME_PVC, - ); - Some(vec![pvc]) - } else { - None - }; + let group_listener_name = validated.node_config.group_listener_name.clone(); + let pvc = listener_operator_volume_source_builder_build_pvc( + &ListenerReference::Listener(group_listener_name), + &unversioned_recommended_labels, + &super::LISTENER_VOLUME_NAME_PVC, + ); + let pvcs = Some(vec![pvc]); pb.add_container(superset_cb.build()); if let Some(termination_grace_period) = merged_config.graceful_shutdown_timeout { @@ -415,7 +408,7 @@ mod tests { }; use super::build_node_rolegroup_statefulset; - use crate::{controller::build::test_support::validated_cluster, crd::SupersetRole}; + use crate::controller::build::test_support::validated_cluster; /// The user-supplied `envOverrides` must be merged in after all operator-set environment /// variables, so that they can override any of them. `CONTAINERDEBUG_LOG_DIRECTORY` is used @@ -425,9 +418,8 @@ mod tests { let cluster = validated_cluster(); let role_group_name = RoleGroupName::from_str("default").expect("valid role group name"); let mut rg = cluster - .role_groups - .get(&SupersetRole::Node) - .and_then(|groups| groups.get(&role_group_name)) + .node_role_group_configs + .get(&role_group_name) .expect("node default role group") .clone(); rg.env_overrides = EnvVarSet::new().with_value( @@ -435,9 +427,8 @@ mod tests { "/stackable/log/user-override", ); - let stateful_set = - build_node_rolegroup_statefulset(&cluster, &SupersetRole::Node, &role_group_name, &rg) - .expect("statefulset built"); + let stateful_set = build_node_rolegroup_statefulset(&cluster, &role_group_name, &rg) + .expect("statefulset built"); let containers = stateful_set .spec diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index a966a623..357d6c22 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -11,7 +11,6 @@ use stackable_operator::{ config::fragment, kube::ResourceExt, product_logging::spec::Logging, - role_utils::GenericRoleConfig, v2::{ controller_utils::{get_cluster_name, get_namespace, get_uid}, product_logging::framework::{ @@ -24,14 +23,13 @@ use stackable_operator::{ }, }, }; -use strum::IntoEnumIterator; use crate::{ built_info::PKG_VERSION_SEMVER, controller::{ CONTAINER_IMAGE_BASE_NAME, SupersetRoleGroupConfig, ValidatedCluster, - ValidatedClusterConfig, ValidatedLogging, ValidatedRoleConfig, ValidatedSupersetConfig, - dereference::DereferencedObjects, + ValidatedClusterConfig, ValidatedClusterParams, ValidatedLogging, ValidatedNodeRoleConfig, + ValidatedRoleConfig, ValidatedSupersetConfig, dereference::DereferencedObjects, }, crd::{ SupersetRole, SupersetRoleGroupType, SupersetRoleType, @@ -155,48 +153,46 @@ pub fn validate_cluster( let cluster_name = get_cluster_name(superset).context(ResolveClusterNameSnafu)?; - let mut role_groups = BTreeMap::new(); - let mut role_configs = BTreeMap::new(); - - for role in SupersetRole::iter() { - let Some(resolved_role) = superset.get_role(&role) else { - continue; - }; - - role_configs.insert( - role.clone(), - ValidatedRoleConfig { - pdb: superset.generic_role_config(&role).map( - |GenericRoleConfig { - pod_disruption_budget, - }| pod_disruption_budget, - ), - listener_class: role.listener_class_name(superset), - group_listener_name: role.group_listener_name(&cluster_name), - }, - ); - - let default_config = SupersetConfig::default_config(&superset.name_any(), &role); + let node_config = ValidatedNodeRoleConfig { + pdb: superset + .spec + .nodes + .role_config + .common + .pod_disruption_budget + .clone(), + listener_class: superset.spec.nodes.role_config.listener_class.clone(), + group_listener_name: SupersetRole::Node + .group_listener_name(&cluster_name) + .expect("The Node role always has a group listener"), + }; + let node_role_group_configs = validate_role_groups( + superset, + &SupersetRole::Node, + &vector_aggregator_config_map_name, + )?; - let mut group_configs = BTreeMap::new(); - for (rolegroup_name, rolegroup) in &resolved_role.role_groups { - let role_group_name = RoleGroupName::from_str(rolegroup_name).with_context(|_| { - ParseRoleGroupNameSnafu { - role_group: rolegroup_name.clone(), - } - })?; - let validated_rg = validate_role_group_config( - &role_group_name, - rolegroup, - resolved_role, - &default_config, - &vector_aggregator_config_map_name, - )?; - group_configs.insert(role_group_name, validated_rg); - } + let worker_config = superset + .spec + .workers + .as_ref() + .map(|workers| ValidatedRoleConfig { + pdb: workers.role_config.common.pod_disruption_budget.clone(), + }); + let worker_role_group_configs = validate_role_groups( + superset, + &SupersetRole::Worker, + &vector_aggregator_config_map_name, + )?; - role_groups.insert(role, group_configs); - } + let beat_config = superset.spec.beat.as_ref().map(|beat| ValidatedRoleConfig { + pdb: beat.role_config.common.pod_disruption_budget.clone(), + }); + let beat_role_group_configs = validate_role_groups( + superset, + &SupersetRole::Beat, + &vector_aggregator_config_map_name, + )?; let cluster_config = &superset.spec.cluster_config; @@ -209,12 +205,12 @@ pub fn validate_cluster( }) }; - Ok(ValidatedCluster::new( - cluster_name, + Ok(ValidatedCluster::new(ValidatedClusterParams { + name: cluster_name, namespace, uid, - resolved_product_image, - ValidatedClusterConfig { + image: resolved_product_image, + cluster_config: ValidatedClusterConfig { authentication_config, opa_config, credentials_secret_name: parse_secret_name(&cluster_config.credentials_secret_name)?, @@ -228,9 +224,45 @@ pub fn validate_cluster( celery_results_backend: cluster_config.celery_results_backend.clone(), celery_broker: cluster_config.celery_broker.clone(), }, - role_groups, - role_configs, - )) + node_config, + node_role_group_configs, + worker_config, + worker_role_group_configs, + beat_config, + beat_role_group_configs, + })) +} + +/// The validated config of every role group of `role`, or an empty map if the role is absent. +fn validate_role_groups( + superset: &SupersetCluster, + role: &SupersetRole, + vector_aggregator_config_map_name: &Option, +) -> Result, Error> { + let Some(resolved_role) = superset.get_role(role) else { + return Ok(BTreeMap::new()); + }; + let default_config = SupersetConfig::default_config(&superset.name_any(), role); + + resolved_role + .role_groups + .iter() + .map(|(rolegroup_name, rolegroup)| { + let role_group_name = RoleGroupName::from_str(rolegroup_name).with_context(|_| { + ParseRoleGroupNameSnafu { + role_group: rolegroup_name.clone(), + } + })?; + let validated = validate_role_group_config( + &role_group_name, + rolegroup, + resolved_role, + &default_config, + vector_aggregator_config_map_name, + )?; + Ok((role_group_name, validated)) + }) + .collect() } /// Merges and validates one role group into a [`SupersetRoleGroupConfig`]. @@ -282,10 +314,7 @@ mod tests { }; use super::{Error, validate_cluster, validate_logging}; - use crate::{ - controller::test_support::default_dereferenced, - crd::{SupersetRole, v1alpha1}, - }; + use crate::{controller::test_support::default_dereferenced, crd::v1alpha1}; /// Builds a [`Logging`] with automatic log configuration for the Superset and Vector containers. fn automatic_logging(enable_vector_agent: bool) -> Logging { @@ -372,9 +401,8 @@ mod tests { let validated = validate_cluster(&superset, dereferenced, "test-repo").expect("validated"); let node = validated - .role_groups - .get(&SupersetRole::Node) - .and_then(|groups| groups.get(&"default".parse().expect("valid role group name"))) + .node_role_group_configs + .get(&"default".parse().expect("valid role group name")) .expect("node default rolegroup"); let overrides = &node.config_overrides.superset_config_py.overrides; diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 639aecb0..d1153b32 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -413,16 +413,6 @@ impl Deref for SupersetRole { } impl SupersetRole { - pub fn listener_class_name( - &self, - superset: &v1alpha1::SupersetCluster, - ) -> Option { - match self { - Self::Node => Some(superset.spec.nodes.role_config.listener_class.clone()), - Self::Worker | Self::Beat => None, - } - } - /// The name of the group listener provided for the role, if the role serves the web UI. /// Nodes will use this group listener so that only one load balancer is needed for that role. /// @@ -605,14 +595,6 @@ impl v1alpha1::SupersetCluster { &self.spec.cluster_config.metadata_database } - pub fn generic_role_config(&self, role: &SupersetRole) -> Option { - self.get_role_config(role).map(|r| r.common.to_owned()) - } - - pub fn get_role_config(&self, role: &SupersetRole) -> Option<&SupersetRoleConfig> { - self.get_role(role).as_ref().map(|c| &c.role_config) - } - pub fn get_role(&self, role: &SupersetRole) -> Option<&SupersetRoleType> { match role { // The `nodes` role is required by the CRD; `Option` is kept for the signature shared