add replicas and pod_overrides to ValidatedRoleGroupConfig

Author: adwk67

rust/operator-binary/src/airflow_controller.rs

@@ -5,7 +5,7 @@ use std::{
 };
 
 use const_format::concatcp;
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::{
         self,
@@ -113,9 +113,6 @@ pub struct Ctx {
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
 pub enum Error {
-    #[snafu(display("object defines no airflow config role"))]
-    NoAirflowRole,
-
     #[snafu(display("failed to apply Service for {rolegroup}"))]
     ApplyRoleGroupService {
         source: stackable_operator::cluster_resources::Error,
@@ -740,11 +737,6 @@ fn build_server_rolegroup_statefulset(
     let authorization_config = &validated_cluster.authorization_config;
     let executor = &validated_cluster.executor;
 
-    let binding = airflow.get_role(airflow_role);
-    let role = binding.as_ref().context(NoAirflowRoleSnafu)?;
-
-    let rolegroup = role.role_groups.get(&rolegroup_ref.role_group);
-
     let mut pb = PodBuilder::new();
     let recommended_object_labels = build_recommended_labels(
         airflow,
@@ -868,7 +860,11 @@ fn build_server_rolegroup_statefulset(
 
     let mut pvcs: Option<Vec<PersistentVolumeClaim>> = None;
 
-    if let Some(listener_group_name) = airflow.group_listener_name(airflow_role) {
+    if let Some(listener_group_name) = validated_cluster
+        .role_configs
+        .get(airflow_role)
+        .and_then(|role_config| role_config.group_listener_name.clone())
+    {
         // Listener endpoints for the Webserver role will use persistent volumes
         // so that load balancers can hard-code the target addresses. This will
         // be the case even when no class is set (and the value defaults to
@@ -981,10 +977,7 @@ fn build_server_rolegroup_statefulset(
         }
     }
     let mut pod_template = pb.build_template();
-    pod_template.merge_from(role.config.pod_overrides.clone());
-    if let Some(rolegroup) = rolegroup {
-        pod_template.merge_from(rolegroup.config.pod_overrides.clone());
-    }
+    pod_template.merge_from(validated_rg_config.pod_overrides.clone());
 
     let restarter_label =
         Label::try_from(("restarter.stackable.tech/enabled", "true")).context(BuildLabelSnafu)?;
@@ -1018,7 +1011,7 @@ fn build_server_rolegroup_statefulset(
             }
             .to_string(),
         ),
-        replicas: rolegroup.and_then(|rg| rg.replicas).map(i32::from),
+        replicas: validated_rg_config.replicas.map(i32::from),
         selector: LabelSelector {
             match_labels: Some(statefulset_match_labels.into()),
             ..LabelSelector::default()

rust/operator-binary/src/controller/validate.rs

@@ -4,6 +4,7 @@ use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection::{self, ResolvedProductImage},
     config::fragment,
+    k8s_openapi::api::core::v1::PodTemplateSpec,
     kube::ResourceExt,
     role_utils::{GenericRoleConfig, RoleGroup},
     v2::role_utils::{GenericCommonConfig, with_validated_config},
@@ -52,6 +53,8 @@ pub struct ValidatedRoleGroupConfig {
     pub merged_config: AirflowConfig,
     pub config_overrides: AirflowConfigOverrides,
     pub env_overrides: HashMap<String, String>,
+    pub replicas: Option<u16>,
+    pub pod_overrides: PodTemplateSpec,
 }
 
 /// The validated cluster: proves that config merging succeeded for every role and
@@ -160,6 +163,8 @@ fn validate_role_group(
         merged_config: validated.config.config,
         config_overrides: validated.config.config_overrides,
         env_overrides: validated.config.env_overrides,
+        replicas: validated.replicas,
+        pod_overrides: validated.config.pod_overrides,
     })
 }
 
@@ -367,4 +372,78 @@ mod tests {
         );
         assert!(validated.env_overrides.is_empty());
     }
+
+    /// `replicas` and the role←role-group merged `pod_overrides` are produced by
+    /// `with_validated_config` and must be carried on `ValidatedRoleGroupConfig`, so the build
+    /// step reads them from here rather than re-deriving from the raw cluster.
+    #[test]
+    fn role_group_carries_merged_pod_overrides_and_replicas() {
+        let cluster_yaml = r#"
+        apiVersion: airflow.stackable.tech/v1alpha2
+        kind: AirflowCluster
+        metadata:
+          name: airflow
+        spec:
+          image:
+            productVersion: 3.1.6
+          clusterConfig:
+            loadExamples: false
+            exposeConfig: false
+            credentialsSecretName: airflow-admin-credentials
+            metadataDatabase:
+              postgresql:
+                host: airflow-postgresql
+                database: airflow
+                credentialsSecretName: airflow-postgresql-credentials
+          webservers:
+            config: {}
+            podOverrides:
+              metadata:
+                labels:
+                  role-label: role
+                  shared: role
+            roleGroups:
+              default:
+                replicas: 3
+                config: {}
+                podOverrides:
+                  metadata:
+                    labels:
+                      rg-label: rg
+                      shared: rg
+          schedulers:
+            config: {}
+            roleGroups:
+              default:
+                config: {}
+          kubernetesExecutors:
+            config: {}
+        "#;
+        let deserializer = serde_yaml::Deserializer::from_str(cluster_yaml);
+        let cluster: v1alpha2::AirflowCluster =
+            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
+        let role = cluster
+            .get_role(&AirflowRole::Webserver)
+            .expect("webserver role");
+        let default_config = AirflowConfig::default_config("airflow", &AirflowRole::Webserver);
+        let rolegroup = role.role_groups.get("default").expect("default role group");
+
+        let validated =
+            validate_role_group(&role, rolegroup, &default_config).expect("validated role group");
+
+        // replicas is carried through from the role group.
+        assert_eq!(validated.replicas, Some(3));
+
+        // pod_overrides is merged role←role-group (role-group wins on shared keys, both levels'
+        // unique keys survive).
+        let labels = validated
+            .pod_overrides
+            .metadata
+            .expect("pod override metadata")
+            .labels
+            .expect("pod override labels");
+        assert_eq!(labels.get("role-label"), Some(&"role".to_string()));
+        assert_eq!(labels.get("rg-label"), Some(&"rg".to_string()));
+        assert_eq!(labels.get("shared"), Some(&"rg".to_string()));
+    }
 }