connect: use persistent volumes for listener and move from deplument to stateful set

Decided to move to stateful set because deployments don't support volume
claim templates.

Author: razvan

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

@@ -11,8 +11,8 @@ use stackable_operator::{
     },
     logging::controller::ReconcilerError,
     status::condition::{
-        compute_conditions, deployment::DeploymentConditionBuilder,
-        operations::ClusterOperationsConditionBuilder,
+        compute_conditions, operations::ClusterOperationsConditionBuilder,
+        statefulset::StatefulSetConditionBuilder,
     },
     time::Duration,
 };
@@ -264,7 +264,7 @@ pub async fn reconcile(
         })?;
 
     let args = server::command_args(&scs.spec.args);
-    let deployment = server::build_deployment(
+    let deployment = server::build_stateful_set(
         scs,
         &server_config,
         &resolved_product_image,
@@ -274,7 +274,7 @@ pub async fn reconcile(
     )
     .context(BuildServerDeploymentSnafu)?;
 
-    let mut ss_cond_builder = DeploymentConditionBuilder::default();
+    let mut ss_cond_builder = StatefulSetConditionBuilder::default();
 
     ss_cond_builder.add(
         cluster_resources

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

@@ -7,15 +7,20 @@ use stackable_operator::{
         configmap::ConfigMapBuilder,
         meta::ObjectMetaBuilder,
         pod::{
-            PodBuilder, container::ContainerBuilder, resources::ResourceRequirementsBuilder,
-            volume::VolumeBuilder,
+            PodBuilder,
+            container::ContainerBuilder,
+            resources::ResourceRequirementsBuilder,
+            volume::{
+                ListenerOperatorVolumeSourceBuilder, ListenerOperatorVolumeSourceBuilderError,
+                ListenerReference, VolumeBuilder,
+            },
         },
     },
     commons::product_image_selection::ResolvedProductImage,
     k8s_openapi::{
         DeepMerge,
         api::{
-            apps::v1::{Deployment, DeploymentSpec},
+            apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
                 ConfigMap, EnvVar, HTTPGetAction, PodSecurityContext, Probe, Service,
                 ServiceAccount, ServicePort, ServiceSpec,
@@ -53,6 +58,11 @@ const HTTP: &str = "http";
 #[derive(Snafu, Debug)]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
+    #[snafu(display("failed to build listener volume"))]
+    BuildListenerVolume {
+        source: ListenerOperatorVolumeSourceBuilderError,
+    },
+
     #[snafu(display("vector agent is enabled but vector aggregator ConfigMap is missing"))]
     VectorAggregatorConfigMapMissing,
 
@@ -188,15 +198,14 @@ pub(crate) fn server_config_map(
         .context(InvalidConfigMapSnafu { name: cm_name })
 }
 
-#[allow(clippy::result_large_err)]
-pub(crate) fn build_deployment(
+pub(crate) fn build_stateful_set(
     scs: &v1alpha1::SparkConnectServer,
     config: &v1alpha1::ServerConfig,
     resolved_product_image: &ResolvedProductImage,
     service_account: &ServiceAccount,
     config_map: &ConfigMap,
     args: Vec<String>,
-) -> Result<Deployment, Error> {
+) -> Result<StatefulSet, Error> {
     let server_role = SparkConnectRole::Server.to_string();
     let recommended_object_labels =
         common::labels(scs, &resolved_product_image.app_version_label, &server_role);
@@ -230,12 +239,6 @@ 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),
@@ -320,13 +323,37 @@ pub(crate) fn build_deployment(
         }
     }
 
+    // Add listener volume
+    let listener_class = &scs.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()),
+            &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
+    };
+
     // 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()) {
         pod_template.merge_from(pod_overrides_spec);
     }
 
-    Ok(Deployment {
+    Ok(StatefulSet {
         metadata: ObjectMetaBuilder::new()
             .name_and_namespace(scs)
             .name(object_name(&scs.name_any(), SparkConnectRole::Server))
@@ -339,9 +366,10 @@ pub(crate) fn build_deployment(
             ))
             .context(MetadataBuildSnafu)?
             .build(),
-        spec: Some(DeploymentSpec {
+        spec: Some(StatefulSetSpec {
             template: pod_template,
             replicas: Some(1),
+            volume_claim_templates: pvcs,
             selector: LabelSelector {
                 match_labels: Some(
                     Labels::role_group_selector(
@@ -355,9 +383,9 @@ pub(crate) fn build_deployment(
                 ),
                 ..LabelSelector::default()
             },
-            ..DeploymentSpec::default()
+            ..StatefulSetSpec::default()
         }),
-        ..Deployment::default()
+        ..StatefulSet::default()
     })
 }
 

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

@@ -23,3 +23,13 @@ pub enum SupportedListenerClasses {
     #[strum(serialize = "external-stable")]
     ExternalStable,
 }
+
+impl SupportedListenerClasses {
+    pub fn discoverable(&self) -> bool {
+        match self {
+            SupportedListenerClasses::ClusterInternal => false,
+            SupportedListenerClasses::ExternalUnstable => true,
+            SupportedListenerClasses::ExternalStable => true,
+        }
+    }
+}

tests/templates/kuttl/spark-connect/10-assert.yaml

@@ -4,7 +4,7 @@ kind: TestAssert
 timeout: 300
 ---
 apiVersion: apps/v1
-kind: Deployment
+kind: StatefulSet
 metadata:
   name: spark-connect-server
 status:
@@ -18,6 +18,7 @@ commands:
   # Sleep to prevent the following spark connect app from failing
   # while the spark-connect server is busy setting up the executors.
   - script: |
+      set -xou pipefail
       sleep 10
       EXECUTOR_COUNT=$(kubectl get pods -n $NAMESPACE --selector 'spark-app-name=spark-connect-server' --field-selector='status.phase=Running' -o NAME|wc -l)
       test 1 -eq "$EXECUTOR_COUNT"