chore: Listener class as webserver role level only

Author: adwk67

deploy/helm/airflow-operator/crds/crds.yaml

@@ -1332,10 +1332,6 @@ spec:
                           description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                           nullable: true
                           type: string
-                        listenerClass:
-                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
-                          nullable: true
-                          type: string
                         logging:
                           default:
                             containers: {}
@@ -1480,11 +1476,16 @@ spec:
                       x-kubernetes-preserve-unknown-fields: true
                     roleConfig:
                       default:
+                        listenerClass: cluster-internal
                         podDisruptionBudget:
                           enabled: true
                           maxUnavailable: null
                       description: This is a product-agnostic RoleConfig, which is sufficient for most of the products.
                       properties:
+                        listenerClass:
+                          default: cluster-internal
+                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+                          type: string
                         podDisruptionBudget:
                           default:
                             enabled: true
@@ -1553,10 +1554,6 @@ spec:
                                 description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                                 nullable: true
                                 type: string
-                              listenerClass:
-                                description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
-                                nullable: true
-                                type: string
                               logging:
                                 default:
                                   containers: {}

docs/modules/airflow/pages/usage-guide/listenerclass.adoc

@@ -8,12 +8,14 @@ The listener defaults to only being accessible from within the Kubernetes cluste
 ----
 spec:
   webservers:
-    config:
+    roleConfig:
       listenerClass: external-unstable  # <1>
+    config:
+      ...
   schedulers:
     ...
   celeryExecutors:
     ...
 ----
-<1> Specify a ListenerClass, such as `external-stable`, `external-unstable`, or `cluster-internal` (the default setting is `cluster-internal`).
+<1> Specify a ListenerClass, such as `external-stable`, `external-unstable`, or `cluster-internal` (the default setting is `cluster-internal`) at role-level.
 This can be set only for the webservers role.

rust/operator-binary/src/airflow_controller.rs

@@ -53,7 +53,7 @@ use stackable_operator::{
         core::{DeserializeGuard, error_boundary},
         runtime::{controller::Action, reflector::ObjectRef},
     },
-    kvp::{Annotation, Label, LabelError, Labels},
+    kvp::{Annotation, Label, LabelError, Labels, ObjectLabels},
     logging::controller::ReconcilerError,
     product_config_utils::{
         CONFIG_OVERRIDE_FILE_FOOTER_KEY, CONFIG_OVERRIDE_FILE_HEADER_KEY, env_vars_from,
@@ -512,26 +512,6 @@ pub async fn reconcile_airflow(
                 &git_sync_resources,
             )?;
 
-            if let Some(listener_class) =
-                airflow.merged_listener_class(&airflow_role, &rolegroup.role_group)
-            {
-                if let Some(listener_group_name) =
-                    airflow.group_listener_name(&airflow_role, &rolegroup)
-                {
-                    let rg_group_listener = build_group_listener(
-                        airflow,
-                        &resolved_product_image,
-                        &rolegroup,
-                        listener_class.to_string(),
-                        listener_group_name,
-                    )?;
-                    cluster_resources
-                        .add(client, rg_group_listener)
-                        .await
-                        .context(ApplyGroupListenerSnafu)?;
-                }
-            }
-
             ss_cond_builder.add(
                 cluster_resources
                     .add(client, rg_statefulset)
@@ -568,6 +548,27 @@ pub async fn reconcile_airflow(
                 .await
                 .context(FailedToCreatePdbSnafu)?;
         }
+
+        if let Some(listener_class) = airflow_role.listener_class_name(airflow) {
+            if let Some(listener_group_name) = airflow.group_listener_name(&airflow_role) {
+                let rg_group_listener = build_group_listener(
+                    airflow,
+                    build_recommended_labels(
+                        airflow,
+                        AIRFLOW_CONTROLLER_NAME,
+                        &resolved_product_image.app_version_label,
+                        role_name,
+                        "none",
+                    ),
+                    listener_class.to_string(),
+                    listener_group_name,
+                )?;
+                cluster_resources
+                    .add(client, rg_group_listener)
+                    .await
+                    .context(ApplyGroupListenerSnafu)?;
+            }
+        }
     }
 
     cluster_resources
@@ -840,8 +841,7 @@ fn build_rolegroup_metadata(
 
 pub fn build_group_listener(
     airflow: &v1alpha1::AirflowCluster,
-    resolved_product_image: &ResolvedProductImage,
-    rolegroup: &RoleGroupRef<v1alpha1::AirflowCluster>,
+    object_labels: ObjectLabels<v1alpha1::AirflowCluster>,
     listener_class: String,
     listener_group_name: String,
 ) -> Result<listener::v1alpha1::Listener> {
@@ -851,13 +851,7 @@ pub fn build_group_listener(
             .name(listener_group_name)
             .ownerreference_from_resource(airflow, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(build_recommended_labels(
-                airflow,
-                AIRFLOW_CONTROLLER_NAME,
-                &resolved_product_image.app_version_label,
-                &rolegroup.role,
-                &rolegroup.role_group,
-            ))
+            .with_recommended_labels(object_labels)
             .context(ObjectMetaSnafu)?
             .build(),
         spec: listener::v1alpha1::ListenerSpec {
@@ -916,7 +910,7 @@ fn build_server_rolegroup_statefulset(
         // A version value is required, and we do want to use the "recommended" format for the other desired labels
         "none",
         &rolegroup_ref.role,
-        &rolegroup_ref.role_group,
+        "none",
     ))
     .context(LabelBuildSnafu)?;
 
@@ -1019,7 +1013,7 @@ fn build_server_rolegroup_statefulset(
 
     let mut pvcs: Option<Vec<PersistentVolumeClaim>> = None;
 
-    if let Some(listener_group_name) = airflow.group_listener_name(airflow_role, rolegroup_ref) {
+    if let Some(listener_group_name) = airflow.group_listener_name(airflow_role) {
         // 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

rust/operator-binary/src/crd/mod.rs

@@ -51,6 +51,7 @@ use crate::crd::{
         AirflowAuthenticationClassResolved, AirflowClientAuthenticationDetails,
         AirflowClientAuthenticationDetailsResolved,
     },
+    v1alpha1::WebserverRoleConfig,
 };
 
 pub mod affinity;
@@ -163,6 +164,8 @@ impl FlaskAppConfigOptions for AirflowConfigOptions {
 
 #[versioned(version(name = "v1alpha1"))]
 pub mod versioned {
+    use crate::crd::v1alpha1::WebserverRoleConfig;
+
     /// An Airflow cluster stacklet. This resource is managed by the Stackable operator for Apache Airflow.
     /// Find more information on how to use it and the resources that the operator generates in the
     /// [operator documentation](DOCS_BASE_URL_PLACEHOLDER/airflow/).
@@ -198,7 +201,7 @@ pub mod versioned {
 
         /// The `webserver` role provides the main UI for user interaction.
         #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub webservers: Option<Role<WebserverConfigFragment>>,
+        pub webservers: Option<Role<AirflowConfigFragment, WebserverRoleConfig>>,
 
         /// The `scheduler` is responsible for triggering jobs and persisting their metadata to the backend database.
         /// Jobs are scheduled on the workers/executors.
@@ -258,6 +261,31 @@ pub mod versioned {
         #[schemars(schema_with = "raw_object_list_schema")]
         pub volume_mounts: Vec<VolumeMount>,
     }
+
+    // TODO: move generic version to op-rs?
+    #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct WebserverRoleConfig {
+        #[serde(flatten)]
+        pub common: GenericRoleConfig,
+
+        /// This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+        #[serde(default = "webserver_default_listener_class")]
+        pub listener_class: String,
+    }
+}
+
+impl Default for v1alpha1::WebserverRoleConfig {
+    fn default() -> Self {
+        v1alpha1::WebserverRoleConfig {
+            listener_class: webserver_default_listener_class(),
+            common: Default::default(),
+        }
+    }
+}
+
+fn webserver_default_listener_class() -> String {
+    "cluster-internal".to_string()
 }
 
 #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq, Serialize)]
@@ -277,16 +305,15 @@ impl HasStatusCondition for v1alpha1::AirflowCluster {
 }
 
 impl v1alpha1::AirflowCluster {
-    /// The name of the group-listener provided for a specific role-group.
+    /// The name of the group-listener provided for a specific role.
     /// Webservers will use this group listener so that only one load balancer
-    /// is needed (per role group).
-    pub fn group_listener_name(
-        &self,
-        role: &AirflowRole,
-        rolegroup: &RoleGroupRef<Self>,
-    ) -> Option<String> {
+    /// is needed for that role.
+    pub fn group_listener_name(&self, role: &AirflowRole) -> Option<String> {
         match role {
-            AirflowRole::Webserver => Some(rolegroup.object_name()),
+            AirflowRole::Webserver => Some(format!(
+                "{cluster_name}-{role}",
+                cluster_name = self.name_any()
+            )),
             AirflowRole::Scheduler | AirflowRole::Worker => None,
         }
     }
@@ -381,34 +408,6 @@ impl v1alpha1::AirflowCluster {
         fragment::validate(conf_rolegroup).context(FragmentValidationFailureSnafu)
     }
 
-    pub fn merged_listener_class(
-        &self,
-        role: &AirflowRole,
-        rolegroup_name: &String,
-    ) -> Option<String> {
-        if role == &AirflowRole::Webserver {
-            if let Some(webservers) = self.spec.webservers.as_ref() {
-                let conf_defaults = Some("cluster-internal".to_string());
-                let mut conf_role = webservers.config.config.listener_class.to_owned();
-                let mut conf_rolegroup = webservers
-                    .role_groups
-                    .get(rolegroup_name)
-                    .map(|rg| rg.config.config.listener_class.clone())
-                    .unwrap_or_default();
-
-                conf_role.merge(&conf_defaults);
-                conf_rolegroup.merge(&conf_role);
-
-                tracing::debug!("Merged listener-class: {:?}", conf_rolegroup);
-                conf_rolegroup
-            } else {
-                None
-            }
-        } else {
-            None
-        }
-    }
-
     /// Retrieve and merge resource configs for the executor template
     pub fn merged_executor_config(
         &self,
@@ -436,18 +435,18 @@ impl v1alpha1::AirflowCluster {
 }
 
 fn extract_role_from_webserver_config(
-    fragment: Role<WebserverConfigFragment>,
+    fragment: Role<AirflowConfigFragment, WebserverRoleConfig>,
 ) -> Role<AirflowConfigFragment> {
     Role {
         config: CommonConfiguration {
-            config: fragment.config.config.airflow_config,
+            config: fragment.config.config,
             config_overrides: fragment.config.config_overrides,
             env_overrides: fragment.config.env_overrides,
             cli_overrides: fragment.config.cli_overrides,
             pod_overrides: fragment.config.pod_overrides,
             product_specific_common_config: fragment.config.product_specific_common_config,
         },
-        role_config: fragment.role_config,
+        role_config: fragment.role_config.common,
         role_groups: fragment
             .role_groups
             .into_iter()
@@ -456,7 +455,7 @@ fn extract_role_from_webserver_config(
                     k,
                     RoleGroup {
                         config: CommonConfiguration {
-                            config: v.config.config.airflow_config,
+                            config: v.config.config,
                             config_overrides: v.config.config_overrides,
                             env_overrides: v.config.env_overrides,
                             cli_overrides: v.config.cli_overrides,
@@ -687,6 +686,17 @@ impl AirflowRole {
         }
         roles
     }
+
+    pub fn listener_class_name(&self, airflow: &v1alpha1::AirflowCluster) -> Option<String> {
+        match self {
+            Self::Webserver => airflow
+                .spec
+                .webservers
+                .to_owned()
+                .map(|webserver| webserver.role_config.listener_class),
+            Self::Worker | Self::Scheduler => None,
+        }
+    }
 }
 
 fn container_debug_command() -> String {
@@ -810,29 +820,6 @@ pub struct AirflowConfig {
     pub graceful_shutdown_timeout: Option<Duration>,
 }
 
-#[derive(Clone, Debug, Default, Fragment, JsonSchema, PartialEq)]
-#[fragment_attrs(
-    derive(
-        Clone,
-        Debug,
-        Default,
-        Deserialize,
-        Merge,
-        JsonSchema,
-        PartialEq,
-        Serialize
-    ),
-    serde(rename_all = "camelCase")
-)]
-pub struct WebserverConfig {
-    #[fragment_attrs(serde(default, flatten))]
-    pub airflow_config: AirflowConfig,
-
-    /// This field controls which [ListenerClass](DOCS_BASE_URL_PLACEHOLDER/listener-operator/listenerclass.html) is used to expose the webserver.
-    #[serde(default)]
-    pub listener_class: String,
-}
-
 impl AirflowConfig {
     pub const CREDENTIALS_SECRET_PROPERTY: &'static str = "credentialsSecret";
     pub const GIT_CREDENTIALS_SECRET_PROPERTY: &'static str = "gitCredentialsSecret";