feat!: Implement server role listener

CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file.
   - Use `--file-log-max-files` (or `FILE_LOG_MAX_FILES`) to limit the number of log files kept.
   - Use `--file-log-rotation-period` (or `FILE_LOG_ROTATION_PERIOD`) to configure the frequency of rotation.
   - Use `--console-log-format` (or `CONSOLE_LOG_FORMAT`) to set the format to `plain` (default) or `json`.
+- BREAKING: Add listener support ([#957]).
 
 ### Changed
 
@@ -40,6 +41,7 @@ All notable changes to this project will be documented in this file.
 [#940]: https://github.com/stackabletech/zookeeper-operator/pull/940
 [#942]: https://github.com/stackabletech/zookeeper-operator/pull/942
 [#946]: https://github.com/stackabletech/zookeeper-operator/pull/946
+[#957]: https://github.com/stackabletech/zookeeper-operator/pull/957
 
 ## [25.3.0] - 2025-03-21
 

deploy/helm/zookeeper-operator/crds/crds.yaml

@@ -28,7 +28,6 @@ spec:
                 clusterConfig:
                   default:
                     authentication: []
-                    listenerClass: cluster-internal
                     tls:
                       quorumSecretClass: tls
                       serverSecretClass: tls
@@ -53,20 +52,6 @@ spec:
                           - authenticationClass
                         type: object
                       type: array
-                    listenerClass:
-                      default: cluster-internal
-                      description: |-
-                        This field controls which type of Service the Operator creates for this ZookeeperCluster:
-
-                        * cluster-internal: Use a ClusterIP service
-
-                        * external-unstable: Use a NodePort 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/nightly/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
-                      type: string
                     tls:
                       default:
                         quorumSecretClass: tls
@@ -444,11 +429,16 @@ spec:
                       x-kubernetes-preserve-unknown-fields: true
                     roleConfig:
                       default:
+                        listenerClass: cluster-internal
                         podDisruptionBudget:
                           enabled: true
                           maxUnavailable: null
                       description: This is a product-agnostic RoleConfig, which is sufficient for most of the products.
                       properties:
+                        listenerClass:
+                          default: cluster-internal
+                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the ZooKeeper nodes.
+                          type: string
                         podDisruptionBudget:
                           default:
                             enabled: true

deploy/helm/zookeeper-operator/templates/roles.yaml

@@ -107,6 +107,17 @@ rules:
     verbs:
       - create
       - patch
+  - apiGroups:
+      - listeners.stackable.tech
+    resources:
+      - listeners
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+      - create
+      - delete
   - apiGroups:
       - {{ include "operator.name" . }}.stackable.tech
     resources:

docs/modules/zookeeper/pages/usage_guide/listenerclass.adoc

@@ -1,16 +1,16 @@
 = Service exposition with ListenerClasses
+:description: Configure the ZooKeeper service exposure with listener classes: cluster-internal, external-unstable or external-stable
 
 Apache ZooKeeper offers an API. The Operator deploys a service called `<name>` (where `<name>` is the name of the ZookeeperCluster) through which ZooKeeper can be reached.
 
-This service can have either the `cluster-internal` or `external-unstable` type. `external-stable` is not supported for ZooKeeper at the moment.
-Read more about the types in the xref:concepts:service-exposition.adoc[service exposition] documentation at platform level.
-
-This is how the listener class is configured:
+The operator deploys a xref:listener-operator:listener.adoc[Listener] for the Server pods.
+The listener defaults to only being accessible from within the Kubernetes cluster, but this can be changed by setting `.spec.nodes.config.listenerClass`:
 
 [source,yaml]
 ----
 spec:
-  clusterConfig:
-    listenerClass: cluster-internal  # <1>
+  servers:
+    roleConfig:
+      listenerClass: external-unstable  # <1>
 ----
-<1> The default `cluster-internal` setting.
+<1> Specify one of `external-stable`, `external-unstable`, `cluster-internal` (the default setting is `cluster-internal`).

rust/operator-binary/src/config/jvm.rs

@@ -1,13 +1,15 @@
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     memory::{BinaryMultiple, MemoryQuantity},
-    role_utils::{self, GenericRoleConfig, JavaCommonConfig, JvmArgumentOverrides, Role},
+    role_utils::{self, JavaCommonConfig, JvmArgumentOverrides, Role},
 };
 
 use crate::crd::{
     JVM_SECURITY_PROPERTIES_FILE, LOG4J_CONFIG_FILE, LOGBACK_CONFIG_FILE, LoggingFramework,
     METRICS_PORT, STACKABLE_CONFIG_DIR, STACKABLE_LOG_CONFIG_DIR,
-    v1alpha1::{ZookeeperCluster, ZookeeperConfig, ZookeeperConfigFragment},
+    v1alpha1::{
+        ZookeeperCluster, ZookeeperConfig, ZookeeperConfigFragment, ZookeeperServerRoleConfig,
+    },
 };
 
 const JAVA_HEAP_FACTOR: f32 = 0.8;
@@ -29,7 +31,7 @@ pub enum Error {
 /// All JVM arguments.
 fn construct_jvm_args(
     zk: &ZookeeperCluster,
-    role: &Role<ZookeeperConfigFragment, GenericRoleConfig, JavaCommonConfig>,
+    role: &Role<ZookeeperConfigFragment, ZookeeperServerRoleConfig, JavaCommonConfig>,
     role_group: &str,
 ) -> Result<Vec<String>, Error> {
     let logging_framework = zk.logging_framework();
@@ -63,7 +65,7 @@ fn construct_jvm_args(
 /// [`construct_zk_server_heap_env`]).
 pub fn construct_non_heap_jvm_args(
     zk: &ZookeeperCluster,
-    role: &Role<ZookeeperConfigFragment, GenericRoleConfig, JavaCommonConfig>,
+    role: &Role<ZookeeperConfigFragment, ZookeeperServerRoleConfig, JavaCommonConfig>,
     role_group: &str,
 ) -> Result<String, Error> {
     let mut jvm_args = construct_jvm_args(zk, role, role_group)?;
@@ -99,7 +101,10 @@ fn is_heap_jvm_argument(jvm_argument: &str) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::crd::{ZookeeperRole, v1alpha1::ZookeeperConfig};
+    use crate::crd::{
+        ZookeeperRole,
+        v1alpha1::{ZookeeperConfig, ZookeeperServerRoleConfig},
+    };
 
     #[test]
     fn test_construct_jvm_arguments_defaults() {
@@ -182,7 +187,7 @@ mod tests {
     ) -> (
         ZookeeperCluster,
         ZookeeperConfig,
-        Role<ZookeeperConfigFragment, GenericRoleConfig, JavaCommonConfig>,
+        Role<ZookeeperConfigFragment, ZookeeperServerRoleConfig, JavaCommonConfig>,
         String,
     ) {
         let zookeeper: ZookeeperCluster =

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

@@ -34,7 +34,7 @@ use stackable_operator::{
 };
 use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
 
-use crate::crd::affinity::get_affinity;
+use crate::crd::{affinity::get_affinity, v1alpha1::ZookeeperServerRoleConfig};
 
 pub mod affinity;
 pub mod authentication;
@@ -47,6 +47,7 @@ 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 METRICS_PORT_NAME: &str = "metrics";
 pub const METRICS_PORT: u16 = 9505;
 
 pub const STACKABLE_DATA_DIR: &str = "/stackable/data";
@@ -72,6 +73,7 @@ pub const MAX_PREPARE_LOG_FILE_SIZE: MemoryQuantity = MemoryQuantity {
 pub const DOCKER_IMAGE_BASE_NAME: &str = "zookeeper";
 
 const DEFAULT_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_minutes_unchecked(2);
+pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal";
 
 mod built_info {
     pub const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION");
@@ -139,7 +141,19 @@ pub mod versioned {
 
         // no doc - it's in the struct.
         #[serde(skip_serializing_if = "Option::is_none")]
-        pub servers: Option<Role<ZookeeperConfigFragment, GenericRoleConfig, JavaCommonConfig>>,
+        pub servers:
+            Option<Role<ZookeeperConfigFragment, ZookeeperServerRoleConfig, JavaCommonConfig>>,
+    }
+
+    #[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct ZookeeperServerRoleConfig {
+        #[serde(flatten)]
+        pub common: GenericRoleConfig,
+
+        /// This field controls which [ListenerClass](DOCS_BASE_URL_PLACEHOLDER/listener-operator/listenerclass.html) is used to expose the ZooKeeper nodes.
+        #[serde(default = "default_listener_class")]
+        pub listener_class: String,
     }
 
     #[derive(Clone, Deserialize, Debug, Eq, JsonSchema, PartialEq, Serialize)]
@@ -164,29 +178,6 @@ pub mod versioned {
             skip_serializing_if = "Option::is_none"
         )]
         pub tls: Option<tls::v1alpha1::ZookeeperTls>,
-
-        /// This field controls which type of Service the Operator creates for this ZookeeperCluster:
-        ///
-        /// * cluster-internal: Use a ClusterIP service
-        ///
-        /// * external-unstable: Use a NodePort service
-        ///
-        /// This is a temporary solution with the goal to keep yaml manifests forward compatible.
-        /// In the future, this setting will control which [ListenerClass](DOCS_BASE_URL_PLACEHOLDER/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,
     }
 
     #[derive(Clone, Debug, Default, Fragment, JsonSchema, PartialEq)]
@@ -354,15 +345,18 @@ fn cluster_config_default() -> v1alpha1::ZookeeperClusterConfig {
         authentication: vec![],
         vector_aggregator_config_map_name: None,
         tls: tls::default_zookeeper_tls(),
-        listener_class: v1alpha1::CurrentlySupportedListenerClasses::default(),
     }
 }
 
-impl v1alpha1::CurrentlySupportedListenerClasses {
-    pub fn k8s_service_type(&self) -> String {
-        match self {
-            v1alpha1::CurrentlySupportedListenerClasses::ClusterInternal => "ClusterIP".to_string(),
-            v1alpha1::CurrentlySupportedListenerClasses::ExternalUnstable => "NodePort".to_string(),
+fn default_listener_class() -> String {
+    DEFAULT_LISTENER_CLASS.to_owned()
+}
+
+impl Default for ZookeeperServerRoleConfig {
+    fn default() -> Self {
+        Self {
+            listener_class: default_listener_class(),
+            common: Default::default(),
         }
     }
 }
@@ -494,6 +488,7 @@ impl HasStatusCondition for v1alpha1::ZookeeperCluster {
 }
 
 impl ZookeeperPodRef {
+    // TODO (@NickLarsenNZ): What to do here?
     pub fn fqdn(&self, cluster_info: &KubernetesClusterInfo) -> String {
         format!(
             "{pod_name}.{service_name}.{namespace}.svc.{cluster_domain}",
@@ -529,16 +524,23 @@ impl v1alpha1::ZookeeperCluster {
         }
     }
 
-    /// The name of the role-level load-balanced Kubernetes `Service`
-    pub fn server_role_service_name(&self) -> Option<String> {
+    /// 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 load-balanced Kubernetes `Service`
-    pub fn server_role_service_fqdn(&self, cluster_info: &KubernetesClusterInfo) -> Option<String> {
+    /// The fully-qualified domain name of the role-level [Listener]
+    ///
+    /// [Listener]: stackable_operator::crd::listener::v1alpha1::Listener
+    pub fn server_role_listener_fqdn(
+        &self,
+        cluster_info: &KubernetesClusterInfo,
+    ) -> Option<String> {
         Some(format!(
-            "{role_service_name}.{namespace}.svc.{cluster_domain}",
-            role_service_name = self.server_role_service_name()?,
+            "{role_listener_name}.{namespace}.svc.{cluster_domain}",
+            role_listener_name = self.server_role_listener_name()?,
             namespace = self.metadata.namespace.as_ref()?,
             cluster_domain = cluster_info.cluster_domain
         ))
@@ -548,8 +550,10 @@ impl v1alpha1::ZookeeperCluster {
     pub fn role(
         &self,
         role_variant: &ZookeeperRole,
-    ) -> Result<&Role<v1alpha1::ZookeeperConfigFragment, GenericRoleConfig, JavaCommonConfig>, Error>
-    {
+    ) -> Result<
+        &Role<v1alpha1::ZookeeperConfigFragment, ZookeeperServerRoleConfig, JavaCommonConfig>,
+        Error,
+    > {
         match role_variant {
             ZookeeperRole::Server => self.spec.servers.as_ref(),
         }
@@ -590,7 +594,7 @@ impl v1alpha1::ZookeeperCluster {
         }
     }
 
-    pub fn role_config(&self, role: &ZookeeperRole) -> Option<&GenericRoleConfig> {
+    pub fn role_config(&self, role: &ZookeeperRole) -> Option<&ZookeeperServerRoleConfig> {
         match role {
             ZookeeperRole::Server => self.spec.servers.as_ref().map(|s| &s.role_config),
         }