adapt to op-rs 0.79.0

Author: maltesander

CHANGELOG.md

@@ -7,6 +7,9 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Support version `3.8.0` ([#753]).
+- The operator can now run on Kubernetes clusters using a non-default cluster domain. It should automatically detect the
+  correct domain to use, but you can also use the env var `KUBERNETES_CLUSTER_DOMAIN` to set the domain explicitly
+  or use the helm-chart property `kubernetesClusterDomain` ([#xxx]).
 
 ### Changed
 
@@ -32,6 +35,7 @@ All notable changes to this project will be documented in this file.
 [#741]: https://github.com/stackabletech/kafka-operator/pull/741
 [#750]: https://github.com/stackabletech/kafka-operator/pull/750
 [#753]: https://github.com/stackabletech/kafka-operator/pull/753
+[#xxx]: https://github.com/stackabletech/kafka-operator/pull/xxx
 
 ## [24.7.0] - 2024-07-24
 

Cargo.lock

@@ -22,7 +22,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
 snafu = "0.8"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.78.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.79.0" }
 strum = { version = "0.26", features = ["derive"] }
 tokio = { version = "1.40", features = ["full"] }
 tracing = "0.1"

Cargo.toml

@@ -47,3 +47,8 @@ nodeSelector: {}
 tolerations: []
 
 affinity: {}
+
+# When running on a non-default Kubernetes cluster domain and the auto detection is not working correctly,
+# you can set your custom cluster domain here.
+# See the https://docs.stackable.tech/home/stable/guides/kubernetes-cluster-domain guide for details
+# kubernetesClusterDomain: my-cluster.local

deploy/helm/kafka-operator/values.yaml

@@ -22,8 +22,10 @@ use stackable_operator::{
             PvcConfig, PvcConfigFragment, Resources, ResourcesFragment,
         },
     },
-    config::fragment::{self, Fragment, ValidationError},
-    config::merge::Merge,
+    config::{
+        fragment::{self, Fragment, ValidationError},
+        merge::Merge,
+    },
     k8s_openapi::{
         api::core::v1::PersistentVolumeClaim, apimachinery::pkg::api::resource::Quantity,
     },
@@ -34,6 +36,7 @@ use stackable_operator::{
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
+    utils::cluster_domain::KUBERNETES_CLUSTER_DOMAIN,
 };
 use std::{collections::BTreeMap, str::FromStr};
 use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
@@ -283,9 +286,12 @@ pub struct KafkaPodRef {
 
 impl KafkaPodRef {
     pub fn fqdn(&self) -> String {
+        let cluster_domain = KUBERNETES_CLUSTER_DOMAIN
+            .get()
+            .expect("KUBERNETES_CLUSTER_DOMAIN must first be set by calling initialize_operator");
         format!(
-            "{}.{}.{}.svc.cluster.local",
-            self.pod_name, self.role_group_service_name, self.namespace
+            "{}.{}.{}.svc.{}",
+            self.pod_name, self.role_group_service_name, self.namespace, cluster_domain
         )
     }
 }

rust/crd/src/lib.rs

@@ -1,12 +1,13 @@
-use crate::{KafkaCluster, STACKABLE_LISTENER_BROKER_DIR};
-
-use crate::security::KafkaTlsSecurity;
-use snafu::{OptionExt, Snafu};
-use stackable_operator::kube::ResourceExt;
 use std::collections::BTreeMap;
 use std::fmt::{Display, Formatter};
+
+use snafu::{OptionExt, Snafu};
+use stackable_operator::{kube::ResourceExt, utils::cluster_domain::KUBERNETES_CLUSTER_DOMAIN};
 use strum::{EnumDiscriminants, EnumString};
 
+use crate::security::KafkaTlsSecurity;
+use crate::{KafkaCluster, STACKABLE_LISTENER_BROKER_DIR};
+
 const LISTENER_LOCAL_ADDRESS: &str = "0.0.0.0";
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -197,10 +198,14 @@ fn node_port_cmd(directory: &str, port_name: &str) -> String {
 }
 
 fn pod_fqdn(kafka: &KafkaCluster, object_name: &str) -> Result<String, KafkaListenerError> {
+    let cluster_domain = KUBERNETES_CLUSTER_DOMAIN
+        .get()
+        .expect("KUBERNETES_CLUSTER_DOMAIN must first be set by calling initialize_operator");
     Ok(format!(
-        "$POD_NAME.{}.{}.svc.cluster.local",
+        "$POD_NAME.{}.{}.svc.{}",
         object_name,
-        kafka.namespace().context(ObjectHasNoNamespaceSnafu)?
+        kafka.namespace().context(ObjectHasNoNamespaceSnafu)?,
+        cluster_domain
     ))
 }
 

rust/crd/src/listener.rs

@@ -78,7 +78,7 @@ async fn main() -> anyhow::Result<()> {
                 "deploy/config-spec/properties.yaml",
                 "/etc/stackable/kafka-operator/config-spec/properties.yaml",
             ])?;
-            let client = client::create_client(Some(OPERATOR_NAME.to_string())).await?;
+            let client = client::initialize_operator(Some(OPERATOR_NAME.to_string())).await?;
             create_controller(client, product_config, watch_namespace).await;
         }
     };