add listener.rs and replace usages of CurrentlySupportedListenerClasses

Author: razvan

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

@@ -1124,16 +1124,6 @@ spec:
                   properties:
                     listenerClass:
                       default: cluster-internal
-                      description: |-
-                        This field controls which type of Service the Operator creates for this HistoryServer:
-
-                        * cluster-internal: Use a ClusterIP service
-
-                        * external-unstable: Use a NodePort service
-
-                        * external-stable: Use a LoadBalancer service
-
-                        This is a temporary solution with the goal to keep yaml manifests forward compatible. In the future, this setting will control which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html> will be used to expose the service, and ListenerClass names will stay the same, allowing for a non-breaking change.
                       enum:
                         - cluster-internal
                         - external-unstable
@@ -1880,21 +1870,12 @@ spec:
                   type: array
                 clusterConfig:
                   default:
-                    listenerClass: external-unstable
+                    listenerClass: cluster-internal
                   description: Global Spark Connect server configuration that applies to all roles.
                   properties:
                     listenerClass:
-                      default: external-unstable
-                      description: |-
-                        This field controls which type of Service the Operator creates for this ConnectServer:
-
-                        * cluster-internal: Use a ClusterIP service
-
-                        * external-unstable: Use a NodePort service
-
-                        * external-stable: Use a LoadBalancer service
-
-                        This is a temporary solution with the goal to keep yaml manifests forward compatible. In the future, this setting will control which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html> will be used to expose the service, and ListenerClass names will stay the same, allowing for a non-breaking change.
+                      default: cluster-internal
+                      description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the Spark services.
                       enum:
                         - cluster-internal
                         - external-unstable

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

@@ -192,21 +192,9 @@ pub async fn reconcile(
         .await
         .context(ApplyRoleBindingSnafu)?;
 
-    // Expose connect server to the outside world
-    let service = server::build_service(scs, &resolved_product_image.app_version_label, None)
-        .context(BuildServiceSnafu)?;
-    cluster_resources
-        .add(client, service.clone())
-        .await
-        .context(ApplyServiceSnafu)?;
-
     // Headless service used by executors connect back to the driver
-    let service = server::build_service(
-        scs,
-        &resolved_product_image.app_version_label,
-        Some("None".to_string()),
-    )
-    .context(BuildServiceSnafu)?;
+    let service = server::build_internal_service(scs, &resolved_product_image.app_version_label)
+        .context(BuildServiceSnafu)?;
 
     cluster_resources
         .add(client, service.clone())

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

@@ -33,7 +33,7 @@ use stackable_operator::{
 use strum::{Display, EnumIter};
 
 use super::common::SparkConnectRole;
-use crate::crd::constants::APP_NAME;
+use crate::crd::{constants::APP_NAME, listener::SupportedListenerClasses};
 
 pub const CONNECT_CONTROLLER_NAME: &str = "connect";
 pub const CONNECT_FULL_CONTROLLER_NAME: &str = concatcp!(
@@ -107,19 +107,9 @@ pub mod versioned {
     #[derive(Clone, Deserialize, Debug, Default, Eq, JsonSchema, PartialEq, Serialize)]
     #[serde(rename_all = "camelCase")]
     pub struct SparkConnectServerClusterConfig {
-        /// This field controls which type of Service the Operator creates for this ConnectServer:
-        ///
-        /// * cluster-internal: Use a ClusterIP service
-        ///
-        /// * external-unstable: Use a NodePort service
-        ///
-        /// * external-stable: Use a LoadBalancer service
-        ///
-        /// This is a temporary solution with the goal to keep yaml manifests forward compatible.
-        /// In the future, this setting will control which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html>
-        /// will be used to expose the service, and ListenerClass names will stay the same, allowing for a non-breaking change.
+        /// This field controls which [ListenerClass](DOCS_BASE_URL_PLACEHOLDER/listener-operator/listenerclass.html) is used to expose the Spark services.
         #[serde(default)]
-        pub listener_class: CurrentlySupportedListenerClasses,
+        pub listener_class: SupportedListenerClasses,
     }
 
     #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Fragment)]
@@ -178,29 +168,6 @@ pub mod versioned {
     }
 }
 
-// TODO: Temporary solution until listener-operator is finished
-#[derive(Clone, Debug, Default, Display, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
-#[serde(rename_all = "PascalCase")]
-pub(crate) enum CurrentlySupportedListenerClasses {
-    #[serde(rename = "cluster-internal")]
-    ClusterInternal,
-    #[default]
-    #[serde(rename = "external-unstable")]
-    ExternalUnstable,
-    #[serde(rename = "external-stable")]
-    ExternalStable,
-}
-
-impl CurrentlySupportedListenerClasses {
-    pub fn k8s_service_type(&self) -> String {
-        match self {
-            CurrentlySupportedListenerClasses::ClusterInternal => "ClusterIP".to_string(),
-            CurrentlySupportedListenerClasses::ExternalUnstable => "NodePort".to_string(),
-            CurrentlySupportedListenerClasses::ExternalStable => "LoadBalancer".to_string(),
-        }
-    }
-}
-
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Fragment)]
 #[fragment_attrs(

rust/operator-binary/src/connect/server.rs

@@ -349,35 +349,13 @@ pub(crate) fn build_deployment(
     })
 }
 
-#[allow(clippy::result_large_err)]
-pub(crate) fn build_service(
+// This is the headless driver service used for the internal
+// communication with the executors as recommended by the Spark docs.
+pub(crate) fn build_internal_service(
     scs: &v1alpha1::SparkConnectServer,
     app_version_label: &str,
-    service_cluster_ip: Option<String>,
 ) -> Result<Service, Error> {
-    let (service_name, service_type, publish_not_ready_addresses) = match service_cluster_ip.clone()
-    {
-        Some(_) => (
-            // These are the properties of the headless driver service used for the internal
-            // communication with the executors as recommended by the Spark docs.
-            //
-            // The flag `publish_not_ready_addresses` *must* be `true` to allow for readiness
-            // probes. Without it, the driver runs into a deadlock beacuse the Pod cannot become
-            // "ready" until the Service is "ready" and vice versa.
-            object_name(&scs.name_any(), SparkConnectRole::Server),
-            "ClusterIP".to_string(),
-            Some(true),
-        ),
-        None => (
-            format!(
-                "{}-{}",
-                object_name(&scs.name_any(), SparkConnectRole::Server),
-                SparkConnectRole::Server
-            ),
-            scs.spec.cluster_config.listener_class.k8s_service_type(),
-            Some(false),
-        ),
-    };
+    let service_name = object_name(&scs.name_any(), SparkConnectRole::Server);
 
     let selector = Labels::role_selector(scs, APP_NAME, &SparkConnectRole::Server.to_string())
         .context(LabelBuildSnafu)?
@@ -398,8 +376,8 @@ pub(crate) fn build_service(
             .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
             .build(),
         spec: Some(ServiceSpec {
-            type_: Some(service_type),
-            cluster_ip: service_cluster_ip,
+            type_: Some("ClusterIP".to_string()),
+            cluster_ip: Some("None".to_string()),
             ports: Some(vec![
                 ServicePort {
                     name: Some(String::from(GRPC)),
@@ -413,7 +391,10 @@ pub(crate) fn build_service(
                 },
             ]),
             selector: Some(selector),
-            publish_not_ready_addresses,
+            // The flag `publish_not_ready_addresses` *must* be `true` to allow for readiness
+            // probes. Without it, the driver runs into a deadlock beacuse the Pod cannot become
+            // "ready" until the Service is "ready" and vice versa.
+            publish_not_ready_addresses: Some(true),
             ..ServiceSpec::default()
         }),
         status: None,

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

@@ -32,7 +32,10 @@ use stackable_operator::{
 use strum::{Display, EnumIter};
 
 use crate::{
-    crd::{affinity::history_affinity, constants::*, logdir::ResolvedLogDir},
+    crd::{
+        affinity::history_affinity, constants::*, listener::SupportedListenerClasses,
+        logdir::ResolvedLogDir,
+    },
     history::config::jvm::construct_history_jvm_args,
 };
 
@@ -62,6 +65,7 @@ pub enum Error {
 
 #[versioned(version(name = "v1alpha1"))]
 pub mod versioned {
+
     /// 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](DOCS_BASE_URL_PLACEHOLDER/spark-k8s/usage-guide/history-server).
@@ -103,42 +107,8 @@ pub mod versioned {
     #[derive(Clone, Deserialize, Debug, Default, Eq, JsonSchema, PartialEq, Serialize)]
     #[serde(rename_all = "camelCase")]
     pub struct SparkHistoryServerClusterConfig {
-        /// This field controls which type of Service the Operator creates for this HistoryServer:
-        ///
-        /// * cluster-internal: Use a ClusterIP service
-        ///
-        /// * external-unstable: Use a NodePort service
-        ///
-        /// * external-stable: Use a LoadBalancer service
-        ///
-        /// This is a temporary solution with the goal to keep yaml manifests forward compatible.
-        /// In the future, this setting will control which ListenerClass <https://docs.stackable.tech/home/stable/listener-operator/listenerclass.html>
-        /// will be used to expose the service, and ListenerClass names will stay the same, allowing for a non-breaking change.
         #[serde(default)]
-        pub listener_class: CurrentlySupportedListenerClasses,
-    }
-}
-
-// TODO: Temporary solution until listener-operator is finished
-#[derive(Clone, Debug, Default, Display, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
-#[serde(rename_all = "PascalCase")]
-pub enum CurrentlySupportedListenerClasses {
-    #[default]
-    #[serde(rename = "cluster-internal")]
-    ClusterInternal,
-    #[serde(rename = "external-unstable")]
-    ExternalUnstable,
-    #[serde(rename = "external-stable")]
-    ExternalStable,
-}
-
-impl CurrentlySupportedListenerClasses {
-    pub fn k8s_service_type(&self) -> String {
-        match self {
-            CurrentlySupportedListenerClasses::ClusterInternal => "ClusterIP".to_string(),
-            CurrentlySupportedListenerClasses::ExternalUnstable => "NodePort".to_string(),
-            CurrentlySupportedListenerClasses::ExternalStable => "LoadBalancer".to_string(),
-        }
+        pub listener_class: SupportedListenerClasses,
     }
 }
 

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

@@ -0,0 +1,25 @@
+use serde::{Deserialize, Serialize};
+use stackable_operator::{
+    config::merge::Atomic,
+    schemars::{self, JsonSchema},
+};
+use strum::Display;
+
+impl Atomic for SupportedListenerClasses {}
+
+#[derive(Clone, Debug, Default, Display, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "PascalCase")]
+pub enum SupportedListenerClasses {
+    #[default]
+    #[serde(rename = "cluster-internal")]
+    #[strum(serialize = "cluster-internal")]
+    ClusterInternal,
+
+    #[serde(rename = "external-unstable")]
+    #[strum(serialize = "external-unstable")]
+    ExternalUnstable,
+
+    #[serde(rename = "external-stable")]
+    #[strum(serialize = "external-stable")]
+    ExternalStable,
+}

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

@@ -55,6 +55,7 @@ use crate::{
 pub mod affinity;
 pub mod constants;
 pub mod history;
+pub mod listener;
 pub mod logdir;
 pub mod roles;
 pub mod tlscerts;