spark connect: move listener to role config

Author: maltesander

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

@@ -2146,8 +2146,10 @@ spec:
                       type: string
                   type: object
                 server:
+                  default:
+                    role_config:
+                      listener_class: cluster-internal
                   description: A Spark Connect server definition.
-                  nullable: true
                   properties:
                     cliOverrides:
                       additionalProperties:
@@ -2157,10 +2159,6 @@ spec:
                     config:
                       default: {}
                       properties:
-                        listenerClass:
-                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the Spark services.
-                          nullable: true
-                          type: string
                         logging:
                           default:
                             containers: {}
@@ -2333,6 +2331,15 @@ spec:
                       description: In the `podOverrides` property you can define a [PodTemplateSpec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#podtemplatespec-v1-core) to override any property that can be set on a Kubernetes Pod. Read the [Pod overrides documentation](https://docs.stackable.tech/home/nightly/concepts/overrides#pod-overrides) for more information.
                       type: object
                       x-kubernetes-preserve-unknown-fields: true
+                    role_config:
+                      properties:
+                        listener_class:
+                          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.
+                          type: string
+                      type: object
+                  required:
+                    - role_config
                   type: object
                 vectorAggregatorConfigMapName:
                   description: Name of the Vector aggregator discovery ConfigMap. It must contain the key `ADDRESS` with the address of the Vector aggregator.

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

@@ -163,6 +163,7 @@ pub async fn reconcile(
         .context(InvalidSparkConnectServerSnafu)?;
 
     let server_config = scs.server_config().context(ServerConfigSnafu)?;
+    let server_role_config = &scs.spec.server.role_config;
     let executor_config = scs.executor_config().context(ExecutorConfigSnafu)?;
 
     let client = &ctx.client;
@@ -286,7 +287,7 @@ pub async fn reconcile(
 
     // ========================================
     // Server listener
-    let listener = server::build_listener(scs, &server_config, &resolved_product_image)
+    let listener = server::build_listener(scs, server_role_config, &resolved_product_image)
         .context(BuildListenerSnafu)?;
 
     cluster_resources

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

@@ -93,14 +93,30 @@ pub mod versioned {
         pub vector_aggregator_config_map_name: Option<String>,
 
         /// A Spark Connect server definition.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub server: Option<CommonConfiguration<ServerConfigFragment, JavaCommonConfig>>,
+        #[serde(default)]
+        pub server: ServerConfigWrapper,
 
         /// Spark Connect executor properties.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub executor: Option<CommonConfiguration<ExecutorConfigFragment, JavaCommonConfig>>,
     }
 
+    /// This struct i
+    #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Serialize, Deserialize)]
+    pub struct ServerConfigWrapper {
+        #[serde(flatten)]
+        pub config: Option<CommonConfiguration<ServerConfigFragment, JavaCommonConfig>>,
+
+        pub role_config: ServerRoleConfig,
+    }
+
+    #[derive(Clone, Debug, JsonSchema, PartialEq, Serialize, Deserialize)]
+    pub struct ServerRoleConfig {
+        /// This field controls which [ListenerClass](DOCS_BASE_URL_PLACEHOLDER/listener-operator/listenerclass.html) is used to expose the Spark services.
+        #[serde(default = "default_listener_class")]
+        pub listener_class: String,
+    }
+
     #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Fragment)]
     #[fragment_attrs(
         derive(
@@ -126,10 +142,6 @@ pub mod versioned {
         /// 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>,
-
-        /// 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: String,
     }
 
     #[derive(Clone, Debug, Default, JsonSchema, PartialEq, Fragment)]
@@ -218,7 +230,6 @@ impl v1alpha1::ServerConfig {
             },
             logging: product_logging::spec::default_logging(),
             requested_secret_lifetime: Some(Self::DEFAULT_CONNECT_SECRET_LIFETIME),
-            listener_class: Some("cluster-internal".into()),
         }
     }
 
@@ -248,7 +259,7 @@ impl v1alpha1::SparkConnectServer {
     pub fn server_config(&self) -> Result<v1alpha1::ServerConfig, Error> {
         let defaults = v1alpha1::ServerConfig::default_config();
         fragment::validate(
-            match self.spec.server.as_ref().map(|cc| cc.config.clone()) {
+            match self.spec.server.config.as_ref().map(|cc| cc.config.clone()) {
                 Some(fragment) => {
                     let mut fc = fragment.clone();
                     fc.merge(&defaults);
@@ -276,6 +287,18 @@ impl v1alpha1::SparkConnectServer {
     }
 }
 
+impl Default for v1alpha1::ServerRoleConfig {
+    fn default() -> Self {
+        v1alpha1::ServerRoleConfig {
+            listener_class: default_listener_class(),
+        }
+    }
+}
+
+pub fn default_listener_class() -> String {
+    "cluster-internal".to_string()
+}
+
 #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SparkConnectServerStatus {

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

@@ -148,6 +148,7 @@ pub(crate) fn server_config_map(
     let jvm_sec_props = common::security_properties(
         scs.spec
             .server
+            .config
             .as_ref()
             .and_then(|s| s.config_overrides.get(JVM_SECURITY_PROPERTIES_FILE)),
     )
@@ -158,6 +159,7 @@ pub(crate) fn server_config_map(
     let metrics_props = common::metrics_properties(
         scs.spec
             .server
+            .config
             .as_ref()
             .and_then(|s| s.config_overrides.get(METRICS_PROPERTIES_FILE)),
     )
@@ -253,6 +255,7 @@ pub(crate) fn build_stateful_set(
     let container_env = env(scs
         .spec
         .server
+        .config
         .as_ref()
         .map(|s| s.env_overrides.clone())
         .as_ref())?;
@@ -344,7 +347,13 @@ pub(crate) fn build_stateful_set(
 
     // Merge user defined pod template if available
     let mut pod_template = pb.build_template();
-    if let Some(pod_overrides_spec) = scs.spec.server.as_ref().map(|s| s.pod_overrides.clone()) {
+    if let Some(pod_overrides_spec) = scs
+        .spec
+        .server
+        .config
+        .as_ref()
+        .map(|s| s.pod_overrides.clone())
+    {
         pod_template.merge_from(pod_overrides_spec);
     }
 
@@ -507,6 +516,7 @@ pub(crate) fn server_properties(
     let config_overrides = scs
         .spec
         .server
+        .config
         .as_ref()
         .and_then(|s| s.config_overrides.get(SPARK_DEFAULTS_FILE_NAME));
 
@@ -581,6 +591,7 @@ fn server_jvm_args(
         &jvm_args,
         scs.spec
             .server
+            .config
             .as_ref()
             .map(|s| &s.product_specific_common_config),
     )
@@ -614,11 +625,11 @@ fn dummy_role_group_ref(
 
 pub(crate) fn build_listener(
     scs: &v1alpha1::SparkConnectServer,
-    config: &v1alpha1::ServerConfig,
+    role_config: &v1alpha1::ServerRoleConfig,
     resolved_product_image: &ResolvedProductImage,
 ) -> Result<listener::v1alpha1::Listener, Error> {
     let listener_name = dummy_role_group_ref(scs).object_name();
-    let listener_class = config.listener_class.clone();
+    let listener_class = role_config.listener_class.clone();
     let role = SparkConnectRole::Server.to_string();
     let recommended_object_labels =
         common::labels(scs, &resolved_product_image.app_version_label, &role);