expose spark connect pods via listener classes

Author: razvan

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

@@ -1,5 +1,7 @@
-= History service exposition with listener classes
-:description: Configure the Spark history service exposure with listener classes: cluster-internal, external-unstable, or external-stable.
+= Service exposition with listener classes
+:description: Configure the Spark connect and history services exposure with listener classes: cluster-internal, external-unstable, or external-stable.
+
+== History services
 
 The operator deploys a xref:listener-operator:listener.adoc[Listener] for each spark history pod.
 The default is to only being accessible from within the Kubernetes cluster, but this can be changed by setting `.spec.clusterConfig.listenerClass`:
@@ -17,3 +19,7 @@ spec:
 <1> Specify one of `external-stable`, `external-unstable`, `cluster-internal` (the default setting is `cluster-internal`).
 
 For the example above, the listener operator creates a service named `spark-history-node-default-0-listener` where `spark-history` is the name of the SparkHistoryServer, `node` is the service role (the only service role available for history servers) and `default` is the role group.
+
+== Connect services
+
+Connect pods can be exposed using listener classes in exactly tha same fashion as history servers.

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

@@ -38,10 +38,11 @@ use crate::{
         },
     },
     crd::constants::{
-        APP_NAME, JVM_SECURITY_PROPERTIES_FILE, LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE,
-        METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE, SPARK_DEFAULTS_FILE_NAME, SPARK_UID,
-        VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_NAME_LOG_CONFIG,
-        VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG, VOLUME_MOUNT_PATH_LOG_CONFIG,
+        APP_NAME, JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME,
+        LOG4J2_CONFIG_FILE, MAX_SPARK_LOG_FILES_SIZE, METRICS_PROPERTIES_FILE, POD_TEMPLATE_FILE,
+        SPARK_DEFAULTS_FILE_NAME, SPARK_UID, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
+        VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG,
+        VOLUME_MOUNT_PATH_LOG_CONFIG,
     },
     product_logging,
 };
@@ -196,12 +197,15 @@ pub(crate) fn build_deployment(
     config_map: &ConfigMap,
     args: Vec<String>,
 ) -> Result<Deployment, Error> {
+    let server_role = SparkConnectRole::Server.to_string();
+    let recommended_object_labels =
+        common::labels(scs, &resolved_product_image.app_version_label, &server_role);
+
+    let recommended_labels =
+        Labels::recommended(recommended_object_labels.clone()).context(LabelBuildSnafu)?;
+
     let metadata = ObjectMetaBuilder::new()
-        .with_recommended_labels(common::labels(
-            scs,
-            &resolved_product_image.app_version_label,
-            &SparkConnectRole::Server.to_string(),
-        ))
+        .with_recommended_labels(recommended_object_labels)
         .context(MetadataBuildSnafu)?
         .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
         .build();
@@ -226,6 +230,12 @@ pub(crate) fn build_deployment(
                 .build(),
         )
         .context(AddVolumeSnafu)?
+        .add_listener_volume_by_listener_class(
+            LISTENER_VOLUME_NAME,
+            &scs.spec.cluster_config.listener_class.to_string(),
+            &recommended_labels.clone(),
+        )
+        .context(AddVolumeSnafu)?
         .security_context(PodSecurityContext {
             run_as_user: Some(SPARK_UID),
             run_as_group: Some(0),
@@ -260,6 +270,8 @@ pub(crate) fn build_deployment(
         .context(AddVolumeMountSnafu)?
         .add_volume_mount(VOLUME_MOUNT_NAME_LOG, VOLUME_MOUNT_PATH_LOG)
         .context(AddVolumeMountSnafu)?
+        .add_volume_mount(LISTENER_VOLUME_NAME, LISTENER_VOLUME_DIR)
+        .context(AddVolumeMountSnafu)?
         .readiness_probe(probe())
         .liveness_probe(probe());