Migrate kafka-operator to use listener-operator for exposing brokers

Cargo.toml

@@ -27,5 +27,6 @@ strum = { version = "0.26", features = ["derive"] }
 tokio = { version = "1.39", features = ["full"] }
 tracing = "0.1"
 
-# [patch."https://github.com/stackabletech/operator-rs.git"]
-# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
+[patch."https://github.com/stackabletech/operator-rs.git"]
+# stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
+stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }

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

@@ -66,11 +66,10 @@ spec:
                               nullable: true
                               type: object
                               x-kubernetes-preserve-unknown-fields: true
-                          required:
-                            - nodeAffinity
-                            - podAffinity
-                            - podAntiAffinity
                           type: object
+                        brokerListenerClass:
+                          nullable: true
+                          type: string
                         gracefulShutdownTimeout:
                           description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                           nullable: true
@@ -333,11 +332,10 @@ spec:
                                     nullable: true
                                     type: object
                                     x-kubernetes-preserve-unknown-fields: true
-                                required:
-                                  - nodeAffinity
-                                  - podAffinity
-                                  - podAntiAffinity
                                 type: object
+                              brokerListenerClass:
+                                nullable: true
+                                type: string
                               gracefulShutdownTimeout:
                                 description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                                 nullable: true
@@ -582,6 +580,9 @@ spec:
                             - configMapName
                           type: object
                       type: object
+                    bootstrapListenerClass:
+                      default: cluster-internal
+                      type: string
                     tls:
                       default:
                         internalSecretClass: tls

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

@@ -2,10 +2,4 @@ apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
   name: {{ include "operator.fullname" . }}-kafka-broker-clusterrole
-rules:
-  - apiGroups:
-      - ""
-    resources:
-      - services
-    verbs:
-      - get
+rules: []

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

@@ -113,6 +113,16 @@ rules:
       - get
       - list
       - watch
+  - apiGroups:
+      - listeners.stackable.tech
+    resources:
+      - listeners
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+      - create
   - apiGroups:
       - rbac.authorization.k8s.io
     resources:

rust/crd/src/lib.rs

@@ -52,7 +52,10 @@ pub const KAFKA_HEAP_OPTS: &str = "KAFKA_HEAP_OPTS";
 // server_properties
 pub const LOG_DIRS_VOLUME_NAME: &str = "log-dirs";
 // directories
-pub const STACKABLE_TMP_DIR: &str = "/stackable/tmp";
+pub const LISTENER_BROKER_VOLUME_NAME: &str = "listener-broker";
+pub const LISTENER_BOOTSTRAP_VOLUME_NAME: &str = "listener-bootstrap";
+pub const STACKABLE_LISTENER_BROKER_DIR: &str = "/stackable/listener-broker";
+pub const STACKABLE_LISTENER_BOOTSTRAP_DIR: &str = "/stackable/listener-bootstrap";
 pub const STACKABLE_DATA_DIR: &str = "/stackable/data";
 pub const STACKABLE_CONFIG_DIR: &str = "/stackable/config";
 pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config";
@@ -159,6 +162,15 @@ pub struct KafkaClusterConfig {
     /// here. When using the [Stackable operator for Apache ZooKeeper](DOCS_BASE_URL_PLACEHOLDER/zookeeper/)
     /// to deploy a ZooKeeper cluster, this will simply be the name of your ZookeeperCluster resource.
     pub zookeeper_config_map_name: String,
+
+    #[serde(default = "KafkaClusterConfig::default_bootstrap_listener_class")]
+    pub bootstrap_listener_class: String,
+}
+
+impl KafkaClusterConfig {
+    fn default_bootstrap_listener_class() -> String {
+        "cluster-internal".to_string()
+    }
 }
 
 impl KafkaCluster {
@@ -405,6 +417,8 @@ pub struct KafkaConfig {
     /// Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
     #[fragment_attrs(serde(default))]
     pub graceful_shutdown_timeout: Option<Duration>,
+
+    pub broker_listener_class: String,
 }
 
 impl KafkaConfig {
@@ -430,6 +444,7 @@ impl KafkaConfig {
             },
             affinity: get_affinity(cluster_name, role),
             graceful_shutdown_timeout: Some(DEFAULT_BROKER_GRACEFUL_SHUTDOWN_TIMEOUT),
+            broker_listener_class: Some("cluster-internal".to_string()),
         }
     }
 }

rust/crd/src/listener.rs

@@ -1,4 +1,4 @@
-use crate::{KafkaCluster, STACKABLE_TMP_DIR};
+use crate::{KafkaCluster, STACKABLE_LISTENER_BROKER_DIR};
 
 use crate::security::KafkaTlsSecurity;
 use snafu::{OptionExt, Snafu};
@@ -8,7 +8,6 @@ use std::fmt::{Display, Formatter};
 use strum::{EnumDiscriminants, EnumString};
 
 const LISTENER_LOCAL_ADDRESS: &str = "0.0.0.0";
-const LISTENER_NODE_ADDRESS: &str = "$NODE";
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
 pub enum KafkaListenerError {
@@ -106,8 +105,11 @@ pub fn get_kafka_listener_config(
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::ClientAuth,
-            host: LISTENER_NODE_ADDRESS.to_string(),
-            port: node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+            host: node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+            port: node_port_cmd(
+                STACKABLE_LISTENER_BROKER_DIR,
+                kafka_security.client_port_name(),
+            ),
         });
         listener_security_protocol_map
             .insert(KafkaListenerName::ClientAuth, KafkaListenerProtocol::Ssl);
@@ -120,8 +122,11 @@ pub fn get_kafka_listener_config(
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
-            host: LISTENER_NODE_ADDRESS.to_string(),
-            port: node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+            host: node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+            port: node_port_cmd(
+                STACKABLE_LISTENER_BROKER_DIR,
+                kafka_security.client_port_name(),
+            ),
         });
         listener_security_protocol_map
             .insert(KafkaListenerName::Client, KafkaListenerProtocol::Ssl);
@@ -134,8 +139,11 @@ pub fn get_kafka_listener_config(
         });
         advertised_listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
-            host: LISTENER_NODE_ADDRESS.to_string(),
-            port: node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+            host: node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+            port: node_port_cmd(
+                STACKABLE_LISTENER_BROKER_DIR,
+                kafka_security.client_port_name(),
+            ),
         });
         listener_security_protocol_map
             .insert(KafkaListenerName::Client, KafkaListenerProtocol::Plaintext);
@@ -180,8 +188,12 @@ pub fn get_kafka_listener_config(
     })
 }
 
+fn node_address_cmd(directory: &str) -> String {
+    format!("$(cat {directory}/default-address/address)")
+}
+
 fn node_port_cmd(directory: &str, port_name: &str) -> String {
-    format!("$(cat {directory}/{port_name}_nodeport)")
+    format!("$(cat {directory}/default-address/ports/{port_name})")
 }
 
 fn pod_fqdn(kafka: &KafkaCluster, object_name: &str) -> Result<String, KafkaListenerError> {
@@ -228,7 +240,6 @@ mod tests {
         "#;
         let kafka: KafkaCluster = serde_yaml::from_str(kafka_cluster).expect("illegal test input");
         let kafka_security = KafkaTlsSecurity::new(
-            &kafka,
             ResolvedAuthenticationClasses::new(vec![AuthenticationClass {
                 metadata: ObjectMetaBuilder::new().name("auth-class").build(),
                 spec: AuthenticationClassSpec {
@@ -260,8 +271,11 @@ mod tests {
             format!(
                 "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port}",
                 name = KafkaListenerName::ClientAuth,
-                host = LISTENER_NODE_ADDRESS,
-                port = node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+                host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+                port = node_port_cmd(
+                    STACKABLE_LISTENER_BROKER_DIR,
+                    kafka_security.client_port_name()
+                ),
                 internal_name = KafkaListenerName::Internal,
                 internal_host = pod_fqdn(&kafka, object_name).unwrap(),
                 internal_port = kafka_security.internal_port(),
@@ -295,7 +309,6 @@ mod tests {
         "#;
         let kafka: KafkaCluster = serde_yaml::from_str(input).expect("illegal test input");
         let kafka_security = KafkaTlsSecurity::new(
-            &kafka,
             ResolvedAuthenticationClasses::new(vec![]),
             "tls".to_string(),
             Some("tls".to_string()),
@@ -320,8 +333,11 @@ mod tests {
             format!(
                 "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port}",
                 name = KafkaListenerName::Client,
-                host = LISTENER_NODE_ADDRESS,
-                port = node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+                host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+                port = node_port_cmd(
+                    STACKABLE_LISTENER_BROKER_DIR,
+                    kafka_security.client_port_name()
+                ),
                 internal_name = KafkaListenerName::Internal,
                 internal_host = pod_fqdn(&kafka, object_name).unwrap(),
                 internal_port = kafka_security.internal_port(),
@@ -357,7 +373,6 @@ mod tests {
         "#;
         let kafka: KafkaCluster = serde_yaml::from_str(input).expect("illegal test input");
         let kafka_security = KafkaTlsSecurity::new(
-            &kafka,
             ResolvedAuthenticationClasses::new(vec![]),
             "".to_string(),
             None,
@@ -382,8 +397,11 @@ mod tests {
             format!(
                 "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port}",
                 name = KafkaListenerName::Client,
-                host = LISTENER_NODE_ADDRESS,
-                port = node_port_cmd(STACKABLE_TMP_DIR, kafka_security.client_port_name()),
+                host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+                port = node_port_cmd(
+                    STACKABLE_LISTENER_BROKER_DIR,
+                    kafka_security.client_port_name()
+                ),
                 internal_name = KafkaListenerName::Internal,
                 internal_host = pod_fqdn(&kafka, object_name).unwrap(),
                 internal_port = kafka_security.internal_port(),