feat: Move ZK ports to headless service

Author: NickLarsenNZ

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

@@ -34,7 +34,11 @@ use stackable_operator::{
 };
 use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
 
-use crate::crd::{affinity::get_affinity, v1alpha1::ZookeeperServerRoleConfig};
+use crate::{
+    crd::{affinity::get_affinity, v1alpha1::ZookeeperServerRoleConfig},
+    discovery::build_role_group_headless_service_name,
+    listener::role_listener_name,
+};
 
 pub mod affinity;
 pub mod authentication;
@@ -47,6 +51,9 @@ pub const OPERATOR_NAME: &str = "zookeeper.stackable.tech";
 pub const ZOOKEEPER_PROPERTIES_FILE: &str = "zoo.cfg";
 pub const JVM_SECURITY_PROPERTIES_FILE: &str = "security.properties";
 
+pub const ZOOKEEPER_LEADER_PORT: u16 = 2888;
+pub const ZOOKEEPER_ELECTION_PORT: u16 = 3888;
+
 pub const METRICS_PORT_NAME: &str = "metrics";
 pub const METRICS_PORT: u16 = 9505;
 
@@ -335,7 +342,7 @@ pub enum ZookeeperRole {
 /// Used for service discovery.
 pub struct ZookeeperPodRef {
     pub namespace: String,
-    pub role_group_service_name: String,
+    pub role_group_headless_service_name: String,
     pub pod_name: String,
     pub zookeeper_myid: u16,
 }
@@ -488,12 +495,11 @@ impl HasStatusCondition for v1alpha1::ZookeeperCluster {
 }
 
 impl ZookeeperPodRef {
-    // TODO (@NickLarsenNZ): What to do here?
-    pub fn fqdn(&self, cluster_info: &KubernetesClusterInfo) -> String {
+    pub fn internal_fqdn(&self, cluster_info: &KubernetesClusterInfo) -> String {
         format!(
             "{pod_name}.{service_name}.{namespace}.svc.{cluster_domain}",
             pod_name = self.pod_name,
-            service_name = self.role_group_service_name,
+            service_name = self.role_group_headless_service_name,
             namespace = self.namespace,
             cluster_domain = cluster_info.cluster_domain
         )
@@ -524,13 +530,6 @@ impl v1alpha1::ZookeeperCluster {
         }
     }
 
-    /// The name of the role-level [Listener]
-    ///
-    /// [Listener]: stackable_operator::crd::listener::v1alpha1::Listener
-    pub fn server_role_listener_name(&self) -> Option<String> {
-        self.metadata.name.clone()
-    }
-
     /// The fully-qualified domain name of the role-level [Listener]
     ///
     /// [Listener]: stackable_operator::crd::listener::v1alpha1::Listener
@@ -540,7 +539,7 @@ impl v1alpha1::ZookeeperCluster {
     ) -> Option<String> {
         Some(format!(
             "{role_listener_name}.{namespace}.svc.{cluster_domain}",
-            role_listener_name = self.server_role_listener_name()?,
+            role_listener_name = role_listener_name(self, &ZookeeperRole::Server),
             namespace = self.metadata.namespace.as_ref()?,
             cluster_domain = cluster_info.cluster_domain
         ))
@@ -626,8 +625,10 @@ impl v1alpha1::ZookeeperCluster {
             for i in 0..rolegroup.replicas.unwrap_or(1) {
                 pod_refs.push(ZookeeperPodRef {
                     namespace: ns.clone(),
-                    role_group_service_name: rolegroup_ref.object_name(),
-                    pod_name: format!("{}-{}", rolegroup_ref.object_name(), i),
+                    role_group_headless_service_name: build_role_group_headless_service_name(
+                        rolegroup_ref.object_name(),
+                    ),
+                    pod_name: format!("{role_group}-{i}", role_group = rolegroup_ref.object_name()),
                     zookeeper_myid: i + myid_offset,
                 });
             }

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

@@ -25,7 +25,13 @@ use stackable_operator::{
     time::Duration,
 };
 
-use crate::crd::{authentication, authentication::ResolvedAuthenticationClasses, tls, v1alpha1};
+use crate::{
+    crd::{
+        authentication::{self, ResolvedAuthenticationClasses},
+        tls, v1alpha1,
+    },
+    zk_controller::LISTENER_VOLUME_NAME,
+};
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
@@ -158,7 +164,7 @@ impl ZookeeperSecurity {
                 .add_volume_mount(tls_volume_name, Self::SERVER_TLS_DIR)
                 .context(AddVolumeMountSnafu)?;
             pod_builder
-                .add_volume(Self::create_tls_volume(
+                .add_volume(Self::create_server_tls_volume(
                     tls_volume_name,
                     secret_class,
                     requested_secret_lifetime,
@@ -172,7 +178,7 @@ impl ZookeeperSecurity {
             .add_volume_mount(tls_volume_name, Self::QUORUM_TLS_DIR)
             .context(AddVolumeMountSnafu)?;
         pod_builder
-            .add_volume(Self::create_tls_volume(
+            .add_volume(Self::create_quorum_tls_volume(
                 tls_volume_name,
                 &self.quorum_secret_class,
                 requested_secret_lifetime,
@@ -298,8 +304,34 @@ impl ZookeeperSecurity {
             .or(self.server_secret_class.as_ref())
     }
 
-    /// Creates ephemeral volumes to mount the `SecretClass` into the Pods
-    fn create_tls_volume(
+    /// Creates ephemeral volumes to mount the `SecretClass` with the listener-volume scope into the Pods.
+    ///
+    /// The resulting volume will contain TLS certificates with the FQDN reported in the applicable [ListenerStatus].
+    ///
+    /// [ListenerStatus]: ::stackable_operator::crd::listener::v1alpha1::ListenerStatus
+    fn create_server_tls_volume(
+        volume_name: &str,
+        secret_class_name: &str,
+        requested_secret_lifetime: &Duration,
+    ) -> Result<Volume> {
+        let volume = VolumeBuilder::new(volume_name)
+            .ephemeral(
+                SecretOperatorVolumeSourceBuilder::new(secret_class_name)
+                    .with_listener_volume_scope(LISTENER_VOLUME_NAME)
+                    .with_format(SecretFormat::TlsPkcs12)
+                    .with_auto_tls_cert_lifetime(*requested_secret_lifetime)
+                    .build()
+                    .context(BuildTlsVolumeSnafu { volume_name })?,
+            )
+            .build();
+
+        Ok(volume)
+    }
+
+    /// Creates ephemeral volumes to mount the `SecretClass` with the pod scope into the Pods.
+    ///
+    /// The resulting volume will contain TLS certificates with the FQDN of the Pod in relation to the StatefulSet's headless service.
+    fn create_quorum_tls_volume(
         volume_name: &str,
         secret_class_name: &str,
         requested_secret_lifetime: &Duration,
@@ -308,7 +340,6 @@ impl ZookeeperSecurity {
             .ephemeral(
                 SecretOperatorVolumeSourceBuilder::new(secret_class_name)
                     .with_pod_scope()
-                    .with_node_scope()
                     .with_format(SecretFormat::TlsPkcs12)
                     .with_auto_tls_cert_lifetime(*requested_secret_lifetime)
                     .build()

rust/operator-binary/src/discovery.rs

@@ -46,7 +46,7 @@ pub enum Error {
     },
 
     #[snafu(display("{listener} has no ingress addresses"))]
-    NoListnerAddresses {
+    NoListenerIngressAddresses {
         listener: ObjectRef<listener::v1alpha1::Listener>,
     },
 
@@ -157,8 +157,7 @@ fn listener_addresses(
         .status
         .as_ref()
         .and_then(|listener_status| listener_status.ingress_addresses.as_ref())
-        // TODO (@NickLarsenNZ): Rename error variant
-        .context(NoListnerAddressesSnafu { listener })?
+        .context(NoListenerIngressAddressesSnafu { listener })?
         .iter()
         // Filter the addresses that have the port we are interested in (they likely all have it though)
         .filter_map(|listener_ingress| {
@@ -185,7 +184,13 @@ fn listener_addresses(
 }
 
 // TODO (@NickLarsenNZ): Implement this directly on RoleGroupRef, ie:
-// RoleGroupRef<K: Resource>::metrics_service_name(&self)
-pub fn build_headless_role_group_metrics_service_name(name: String) -> String {
+// RoleGroupRef<K: Resource>::metrics_service_name(&self) to restrict what _name_ can be.
+pub fn build_role_group_headless_service_name(name: String) -> String {
+    format!("{name}-headless")
+}
+
+// TODO (@NickLarsenNZ): Implement this directly on RoleGroupRef, ie:
+// RoleGroupRef<K: Resource>::metrics_service_name(&self) to restrict what _name_ can be.
+pub fn build_role_group_metrics_service_name(name: String) -> String {
     format!("{name}-metrics")
 }

rust/operator-binary/src/listener.rs

@@ -73,7 +73,7 @@ pub fn build_role_listener(
 }
 
 // TODO (@NickLarsenNZ): This could be a method we can put on a Resource that takes a role_name
-fn role_listener_name(zk: &v1alpha1::ZookeeperCluster, zk_role: &ZookeeperRole) -> String {
+pub fn role_listener_name(zk: &v1alpha1::ZookeeperCluster, zk_role: &ZookeeperRole) -> String {
     // TODO (@NickLarsenNZ): Make a convention, do we use name_any() and allow empty string? or metadata.name.expect, or handle the error?
     format!("{zk}-{zk_role}", zk = zk.name_any())
 }

rust/operator-binary/src/zk_controller.rs

@@ -79,12 +79,15 @@ use crate::{
         DOCKER_IMAGE_BASE_NAME, JVM_SECURITY_PROPERTIES_FILE, MAX_PREPARE_LOG_FILE_SIZE,
         MAX_ZK_LOG_FILES_SIZE, METRICS_PORT, METRICS_PORT_NAME, STACKABLE_CONFIG_DIR,
         STACKABLE_DATA_DIR, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR, STACKABLE_RW_CONFIG_DIR,
-        ZOOKEEPER_PROPERTIES_FILE, ZookeeperRole,
+        ZOOKEEPER_ELECTION_PORT, ZOOKEEPER_LEADER_PORT, ZOOKEEPER_PROPERTIES_FILE, ZookeeperRole,
         security::{self, ZookeeperSecurity},
         v1alpha1::{self, ZookeeperServerRoleConfig},
     },
-    discovery::{self, build_discovery_configmap, build_headless_role_group_metrics_service_name},
-    listener::build_role_listener,
+    discovery::{
+        self, build_discovery_configmap, build_role_group_headless_service_name,
+        build_role_group_metrics_service_name,
+    },
+    listener::{build_role_listener, role_listener_name},
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::extend_role_group_config_map,
     utils::build_recommended_labels,
@@ -430,7 +433,10 @@ pub async fn reconcile_zk(
             .merged_config(&ZookeeperRole::Server, &rolegroup)
             .context(FailedToResolveConfigSnafu)?;
 
-        let rg_service = build_server_rolegroup_service(zk, &rolegroup, &resolved_product_image)?;
+        let rg_headless_service =
+            build_server_rolegroup_headless_service(zk, &rolegroup, &resolved_product_image)?;
+        let rg_metrics_service =
+            build_server_rolegroup_metrics_service(zk, &rolegroup, &resolved_product_image)?;
         let rg_configmap = build_server_rolegroup_config_map(
             zk,
             &rolegroup,
@@ -451,7 +457,13 @@ pub async fn reconcile_zk(
         )?;
 
         cluster_resources
-            .add(client, rg_service)
+            .add(client, rg_headless_service)
+            .await
+            .with_context(|_| ApplyRoleGroupServiceSnafu {
+                rolegroup: rolegroup.clone(),
+            })?;
+        cluster_resources
+            .add(client, rg_metrics_service)
             .await
             .with_context(|_| ApplyRoleGroupServiceSnafu {
                 rolegroup: rolegroup.clone(),
@@ -556,11 +568,11 @@ fn build_server_rolegroup_config_map(
         .flatten()
         .map(|pod| {
             (
-                format!("server.{}", pod.zookeeper_myid),
+                format!("server.{id}", id = pod.zookeeper_myid),
                 format!(
-                    "{}:2888:3888;{}",
-                    pod.fqdn(cluster_info),
-                    zookeeper_security.client_port()
+                    "{internal_fqdn}:{ZOOKEEPER_LEADER_PORT}:{ZOOKEEPER_ELECTION_PORT};{client_port}",
+                    internal_fqdn = pod.internal_fqdn(cluster_info),
+                    client_port = zookeeper_security.client_port()
                 ),
             )
         })
@@ -642,10 +654,69 @@ fn build_server_rolegroup_config_map(
         })
 }
 
-/// The rolegroup [`Service`] is a headless service that allows direct access to the instances of a certain rolegroup
+/// The rolegroup [`Service`] is a headless service that allows internal access to the instances of a certain rolegroup
 ///
 /// This is mostly useful for internal communication between peers, or for clients that perform client-side load balancing.
-fn build_server_rolegroup_service(
+fn build_server_rolegroup_headless_service(
+    zk: &v1alpha1::ZookeeperCluster,
+    rolegroup: &RoleGroupRef<v1alpha1::ZookeeperCluster>,
+    resolved_product_image: &ResolvedProductImage,
+) -> Result<Service> {
+    let metadata = ObjectMetaBuilder::new()
+        .name_and_namespace(zk)
+        .name(build_role_group_headless_service_name(
+            rolegroup.object_name(),
+        ))
+        .ownerreference_from_resource(zk, None, Some(true))
+        .context(ObjectMissingMetadataForOwnerRefSnafu)?
+        .with_recommended_labels(build_recommended_labels(
+            zk,
+            ZK_CONTROLLER_NAME,
+            &resolved_product_image.app_version_label,
+            &rolegroup.role,
+            &rolegroup.role_group,
+        ))
+        .context(ObjectMetaSnafu)?
+        .build();
+
+    let service_selector_labels =
+        Labels::role_group_selector(zk, APP_NAME, &rolegroup.role, &rolegroup.role_group)
+            .context(BuildLabelSnafu)?;
+
+    let service_spec = ServiceSpec {
+        // Internal communication does not need to be exposed
+        type_: Some("ClusterIP".to_string()),
+        cluster_ip: Some("None".to_string()),
+        ports: Some(vec![
+            ServicePort {
+                // TODO (@NickLarsenNZ): Use a const
+                name: Some("zk-leader".to_string()),
+                port: ZOOKEEPER_LEADER_PORT as i32,
+                protocol: Some("TCP".to_string()),
+                ..ServicePort::default()
+            },
+            ServicePort {
+                // TODO (@NickLarsenNZ): Use a const
+                name: Some("zk-election".to_string()),
+                port: ZOOKEEPER_ELECTION_PORT as i32,
+                protocol: Some("TCP".to_string()),
+                ..ServicePort::default()
+            },
+        ]),
+        selector: Some(service_selector_labels.into()),
+        publish_not_ready_addresses: Some(true),
+        ..ServiceSpec::default()
+    };
+
+    Ok(Service {
+        metadata,
+        spec: Some(service_spec),
+        status: None,
+    })
+}
+
+/// The rolegroup [`Service`] for exposing metrics
+fn build_server_rolegroup_metrics_service(
     zk: &v1alpha1::ZookeeperCluster,
     rolegroup: &RoleGroupRef<v1alpha1::ZookeeperCluster>,
     resolved_product_image: &ResolvedProductImage,
@@ -655,7 +726,7 @@ fn build_server_rolegroup_service(
 
     let metadata = ObjectMetaBuilder::new()
         .name_and_namespace(zk)
-        .name(build_headless_role_group_metrics_service_name(
+        .name(build_role_group_metrics_service_name(
             rolegroup.object_name(),
         ))
         .ownerreference_from_resource(zk, None, Some(true))
@@ -767,8 +838,7 @@ fn build_server_rolegroup_statefulset(
     // .context(LabelBuildSnafu)?;
 
     let listener_pvc = build_role_listener_pvc(
-        &zk.server_role_listener_name()
-            .expect("todo: get role from zk_role"),
+        &role_listener_name(zk, &ZookeeperRole::Server),
         &unversioned_recommended_labels,
     )?;
 
@@ -901,8 +971,8 @@ fn build_server_rolegroup_statefulset(
             ..Probe::default()
         })
         .add_container_port("zk", zookeeper_security.client_port().into())
-        .add_container_port("zk-leader", 2888)
-        .add_container_port("zk-election", 3888)
+        .add_container_port("zk-leader", ZOOKEEPER_LEADER_PORT as i32)
+        .add_container_port("zk-election", ZOOKEEPER_ELECTION_PORT as i32)
         .add_container_port("metrics", 9505)
         .add_volume_mount("data", STACKABLE_DATA_DIR)
         .context(AddVolumeMountSnafu)?
@@ -1058,7 +1128,7 @@ fn build_server_rolegroup_statefulset(
             match_labels: Some(statefulset_match_labels.into()),
             ..LabelSelector::default()
         },
-        service_name: Some(build_headless_role_group_metrics_service_name(
+        service_name: Some(build_role_group_headless_service_name(
             rolegroup_ref.object_name(),
         )),
         template: pod_template,