history: create group listeners and update crd

Author: razvan

deploy/helm/spark-k8s-operator/crds/crds.yaml

@@ -1118,17 +1118,8 @@ spec:
               description: A Spark cluster history server component. This resource is managed by the Stackable operator for Apache Spark. Find more information on how to use it in the [operator documentation](https://docs.stackable.tech/home/nightly/spark-k8s/usage-guide/history-server).
               properties:
                 clusterConfig:
-                  default:
-                    listenerClass: cluster-internal
+                  default: {}
                   description: Global Spark history server configuration that applies to all roles and role groups.
-                  properties:
-                    listenerClass:
-                      default: cluster-internal
-                      enum:
-                        - cluster-internal
-                        - external-unstable
-                        - external-stable
-                      type: string
                   type: object
                 image:
                   anyOf:
@@ -1385,6 +1376,9 @@ spec:
                         cleaner:
                           nullable: true
                           type: boolean
+                        listenerClass:
+                          nullable: true
+                          type: string
                         logging:
                           default:
                             containers: {}
@@ -1631,6 +1625,9 @@ spec:
                               cleaner:
                                 nullable: true
                                 type: boolean
+                              listenerClass:
+                                nullable: true
+                                type: string
                               logging:
                                 default:
                                   containers: {}

deploy/helm/spark-k8s-operator/templates/roles.yaml

@@ -133,3 +133,14 @@ rules:
       - bind
     resourceNames:
       - {{ include "operator.name" . }}-clusterrole
+  - apiGroups:
+      - listeners.stackable.tech
+    resources:
+      - listeners
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+      - create
+      - delete

docs/modules/spark-k8s/pages/usage-guide/listenerclass.adoc

@@ -13,8 +13,9 @@ kind: SparkHistoryServer
 metadata:
   name: spark-history
 spec:
-  clusterConfig:
-    listenerClass: external-unstable  # <1>
+  nodes:
+    config:
+      listenerClass: external-unstable  # <1>
 ----
 <1> Specify one of `external-stable`, `external-unstable`, `cluster-internal` (the default setting is `cluster-internal`).
 

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

@@ -88,6 +88,7 @@ pub const SPARK_ENV_SH_FILE_NAME: &str = "spark-env.sh";
 pub const SPARK_CLUSTER_ROLE: &str = "spark-k8s-clusterrole";
 pub const SPARK_UID: i64 = 1000;
 pub const METRICS_PORT: u16 = 18081;
+pub const HISTORY_UI_PORT: u16 = 18080;
 
 pub const LISTENER_VOLUME_NAME: &str = "listener";
 pub const LISTENER_VOLUME_DIR: &str = "/stackable/listener";

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

@@ -32,10 +32,7 @@ use stackable_operator::{
 use strum::{Display, EnumIter};
 
 use crate::{
-    crd::{
-        affinity::history_affinity, constants::*, listener::SupportedListenerClasses,
-        logdir::ResolvedLogDir,
-    },
+    crd::{affinity::history_affinity, constants::*, logdir::ResolvedLogDir},
     history::config::jvm::construct_history_jvm_args,
 };
 
@@ -106,13 +103,17 @@ pub mod versioned {
 
     #[derive(Clone, Deserialize, Debug, Default, Eq, JsonSchema, PartialEq, Serialize)]
     #[serde(rename_all = "camelCase")]
-    pub struct SparkHistoryServerClusterConfig {
-        #[serde(default)]
-        pub listener_class: SupportedListenerClasses,
-    }
+    pub struct SparkHistoryServerClusterConfig {}
 }
 
 impl v1alpha1::SparkHistoryServer {
+    /// The name of the group-listener provided for a specific role-group.
+    /// History servers will use this group listener so that only one load balancer
+    /// is needed (per role group).
+    pub fn group_listener_name(&self, rolegroup: &RoleGroupRef<Self>) -> String {
+        format!("{}-group", rolegroup.object_name())
+    }
+
     /// Returns a reference to the role. Raises an error if the role is not defined.
     pub fn role(&self) -> &Role<HistoryConfigFragment, GenericRoleConfig, JavaCommonConfig> {
         &self.spec.nodes
@@ -343,6 +344,9 @@ pub struct HistoryConfig {
     /// This can be shortened by the `maxCertificateLifetime` setting on the SecretClass issuing the TLS certificate.
     #[fragment_attrs(serde(default))]
     pub requested_secret_lifetime: Option<Duration>,
+
+    #[serde(default)]
+    pub listener_class: String,
 }
 
 impl HistoryConfig {
@@ -366,6 +370,7 @@ impl HistoryConfig {
             logging: product_logging::spec::default_logging(),
             affinity: history_affinity(cluster_name),
             requested_secret_lifetime: Some(Self::DEFAULT_HISTORY_SECRET_LIFETIME),
+            listener_class: Some(default_listener_class()),
         }
     }
 }
@@ -402,6 +407,10 @@ impl Configuration for HistoryConfigFragment {
     }
 }
 
+fn default_listener_class() -> String {
+    "cluster-internal".to_string()
+}
+
 #[cfg(test)]
 mod test {
     use indoc::indoc;

rust/operator-binary/src/history/history_controller.rs

@@ -21,7 +21,10 @@ use stackable_operator::{
         },
     },
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
-    commons::product_image_selection::ResolvedProductImage,
+    commons::{
+        listener::{Listener, ListenerPort, ListenerSpec},
+        product_image_selection::ResolvedProductImage,
+    },
     k8s_openapi::{
         DeepMerge,
         api::{
@@ -56,7 +59,7 @@ use crate::{
     Ctx,
     crd::{
         constants::{
-            ACCESS_KEY_ID, APP_NAME, HISTORY_CONTROLLER_NAME, HISTORY_ROLE_NAME,
+            ACCESS_KEY_ID, APP_NAME, HISTORY_CONTROLLER_NAME, HISTORY_ROLE_NAME, HISTORY_UI_PORT,
             JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME,
             MAX_SPARK_LOG_FILES_SIZE, METRICS_PORT, OPERATOR_NAME, SECRET_ACCESS_KEY,
             SPARK_CLUSTER_ROLE, SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME,
@@ -76,6 +79,11 @@ use crate::{
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
+    #[snafu(display("failed to build object meta data"))]
+    ObjectMeta {
+        source: stackable_operator::builder::meta::Error,
+    },
+
     #[snafu(display("failed to build listener volume"))]
     BuildListenerVolume {
         source: ListenerOperatorVolumeSourceBuilderError,
@@ -216,9 +224,12 @@ pub enum Error {
 
     #[snafu(display("failed to merge environment config and/or overrides"))]
     MergeEnv { source: crate::crd::history::Error },
-}
 
-type Result<T, E = Error> = std::result::Result<T, E>;
+    #[snafu(display("failed to apply group listener"))]
+    ApplyGroupListener {
+        source: stackable_operator::cluster_resources::Error,
+    },
+}
 
 impl ReconcilerError for Error {
     fn category(&self) -> &'static str {
@@ -229,7 +240,7 @@ impl ReconcilerError for Error {
 pub async fn reconcile(
     shs: Arc<DeserializeGuard<v1alpha1::SparkHistoryServer>>,
     ctx: Arc<Ctx>,
-) -> Result<Action> {
+) -> Result<Action, Error> {
     tracing::info!("Starting reconcile history server");
 
     let shs = shs
@@ -322,6 +333,17 @@ pub async fn reconcile(
                 .add(client, sts)
                 .await
                 .context(ApplyDeploymentSnafu)?;
+
+            let rg_group_listener = build_group_listener(
+                shs,
+                &resolved_product_image,
+                &rgr,
+                merged_config.listener_class.to_string(),
+            )?;
+            cluster_resources
+                .add(client, rg_group_listener)
+                .await
+                .context(ApplyGroupListenerSnafu)?;
         }
 
         let role_config = &shs.spec.nodes.role_config;
@@ -343,6 +365,50 @@ pub async fn reconcile(
     Ok(Action::await_change())
 }
 
+#[allow(clippy::result_large_err)]
+fn build_group_listener(
+    shs: &v1alpha1::SparkHistoryServer,
+    resolved_product_image: &ResolvedProductImage,
+    rolegroup: &RoleGroupRef<v1alpha1::SparkHistoryServer>,
+    listener_class: String,
+) -> Result<Listener, Error> {
+    Ok(Listener {
+        metadata: ObjectMetaBuilder::new()
+            .name_and_namespace(shs)
+            .name(shs.group_listener_name(rolegroup))
+            .ownerreference_from_resource(shs, None, Some(true))
+            .context(ObjectMissingMetadataForOwnerRefSnafu)?
+            .with_recommended_labels(labels(
+                shs,
+                &resolved_product_image.app_version_label,
+                &rolegroup.role_group,
+            ))
+            .context(ObjectMetaSnafu)?
+            .build(),
+        spec: ListenerSpec {
+            class_name: Some(listener_class),
+            ports: Some(listener_ports()),
+            ..ListenerSpec::default()
+        },
+        status: None,
+    })
+}
+
+fn listener_ports() -> Vec<ListenerPort> {
+    vec![
+        ListenerPort {
+            name: "metrics".to_string(),
+            port: METRICS_PORT.into(),
+            protocol: Some("TCP".to_string()),
+        },
+        ListenerPort {
+            name: "http".to_string(),
+            port: HISTORY_UI_PORT.into(),
+            protocol: Some("TCP".to_string()),
+        },
+    ]
+}
+
 pub fn error_policy(
     _obj: Arc<DeserializeGuard<v1alpha1::SparkHistoryServer>>,
     error: &Error,
@@ -528,7 +594,7 @@ fn build_stateful_set(
             "-c".to_string(),
         ])
         .args(command_args(log_dir))
-        .add_container_port("http", 18080)
+        .add_container_port("http", HISTORY_UI_PORT.into())
         .add_container_port("metrics", METRICS_PORT.into())
         .add_env_vars(merged_env)
         .add_volume_mounts(log_dir.volume_mounts())
@@ -544,28 +610,19 @@ fn build_stateful_set(
         .build();
 
     // Add listener volume
-    let listener_class = &shs.spec.cluster_config.listener_class;
-    let pvcs = if listener_class.discoverable() {
-        // externally reachable listener endpoints will use persistent volumes
-        // so that load balancers can hard-code the target addresses
-        let pvc = ListenerOperatorVolumeSourceBuilder::new(
-            &ListenerReference::ListenerClass(listener_class.to_string()),
+    // 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
+    // cluster-internal) as the address should still be consistent.
+    let pvcs = Some(vec![
+        ListenerOperatorVolumeSourceBuilder::new(
+            &ListenerReference::ListenerClass(merged_config.listener_class.to_string()),
             &recommended_labels,
         )
         .context(BuildListenerVolumeSnafu)?
         .build_pvc(LISTENER_VOLUME_NAME.to_string())
-        .context(BuildListenerVolumeSnafu)?;
-        Some(vec![pvc])
-    } else {
-        // non-reachable endpoints use ephemeral volumes
-        pb.add_listener_volume_by_listener_class(
-            LISTENER_VOLUME_NAME,
-            &listener_class.to_string(),
-            &recommended_labels,
-        )
-        .context(AddVolumeSnafu)?;
-        None
-    };
+        .context(BuildListenerVolumeSnafu)?,
+    ]);
 
     pb.add_container(container);
 
@@ -642,7 +699,7 @@ fn build_stateful_set(
 fn build_history_role_serviceaccount(
     shs: &v1alpha1::SparkHistoryServer,
     app_version_label: &str,
-) -> Result<(ServiceAccount, RoleBinding)> {
+) -> Result<(ServiceAccount, RoleBinding), Error> {
     let sa = ServiceAccount {
         metadata: ObjectMetaBuilder::new()
             .name_and_namespace(shs)
@@ -784,11 +841,11 @@ fn build_rolegroup_service(
     shs: &v1alpha1::SparkHistoryServer,
     app_version_label: &str,
     group: &RoleGroupRef<v1alpha1::SparkHistoryServer>,
-) -> Result<Service> {
+) -> Result<Service, Error> {
     let ports = Some(vec![
         ServicePort {
             name: Some(String::from("http")),
-            port: 18080,
+            port: HISTORY_UI_PORT.into(),
             ..ServicePort::default()
         },
         ServicePort {

tests/templates/kuttl/spark-history-server/06-deploy-history-server.yaml.j2

@@ -36,9 +36,6 @@ spec:
   vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
 
-  clusterConfig:
-    listenerClass: external-unstable
-
   logFileDirectory:
     s3:
       prefix: eventlogs/
@@ -48,6 +45,7 @@ spec:
   #sparkConf:
   nodes:
     config:
+      listenerClass: external-unstable
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
         containers: