thread through name, namespace, uid via ObjectMeta

adwk67 · adwk67 · commit 662e9a14e0d1 · 2026-06-10T17:23:05.000+02:00
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
@@ -1,6 +1,6 @@
 //! Ensures that `Pod`s are configured and running for each [`v1alpha1::KafkaCluster`].
 
-use std::{collections::BTreeMap, sync::Arc};
+use std::{borrow::Cow, collections::BTreeMap, sync::Arc};
 
 use const_format::concatcp;
 use snafu::{ResultExt, Snafu};
@@ -11,7 +11,7 @@ use stackable_operator::{
     crd::listener,
     kube::{
         Resource,
-        api::DynamicObject,
+        api::{DynamicObject, ObjectMeta},
         core::{DeserializeGuard, error_boundary},
         runtime::{controller::Action, reflector::ObjectRef},
     },
@@ -22,6 +22,10 @@ use stackable_operator::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
+    v2::types::{
+        kubernetes::{NamespaceName, Uid},
+        operator::ClusterName,
+    },
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -38,7 +42,6 @@ use crate::{
         security::KafkaTlsSecurity,
         v1alpha1,
     },
-    discovery::{self, build_discovery_configmap},
     operations::pdb::add_pdbs,
     resource::{
         listener::build_broker_rolegroup_bootstrap_listener,
@@ -94,7 +97,7 @@ pub enum Error {
     },
 
     #[snafu(display("failed to build discovery ConfigMap"))]
-    BuildDiscoveryConfig { source: discovery::Error },
+    BuildDiscoveryConfig { source: build::discovery::Error },
 
     #[snafu(display("failed to apply discovery ConfigMap"))]
     ApplyDiscoveryConfig {
@@ -208,7 +211,17 @@ impl ReconcilerError for Error {
 
 /// The validated cluster. Carries everything the build steps need, resolved once
 /// here so downstream code never re-derives it or touches the raw spec.
+///
+/// The cluster identity (`name`, `namespace`, `uid`) is captured here so that owner
+/// references for child objects can be built straight from this struct (via its
+/// [`Resource`] impl) without threading the raw [`v1alpha1::KafkaCluster`] around.
+/// This mirrors the hive-/opensearch-operator's `ValidatedCluster`.
 pub struct ValidatedKafkaCluster {
+    /// `ObjectMeta` carrying `name`, `namespace` and `uid`, so this struct can act as the
+    /// owner [`Resource`] for child objects.
+    metadata: ObjectMeta,
+    pub name: ClusterName,
+    pub namespace: NamespaceName,
     pub image: ResolvedProductImage,
     pub kafka_security: KafkaTlsSecurity,
     // DESIGN DECISION: the dereferenced authorization config is folded into the
@@ -222,6 +235,69 @@ pub struct ValidatedKafkaCluster {
     pub metadata_manager: MetadataManager,
 }
 
+impl ValidatedKafkaCluster {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        name: ClusterName,
+        namespace: NamespaceName,
+        uid: Uid,
+        image: ResolvedProductImage,
+        kafka_security: KafkaTlsSecurity,
+        authorization_config: Option<KafkaAuthorizationConfig>,
+        role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
+        pod_descriptors: Vec<KafkaPodDescriptor>,
+        metadata_manager: MetadataManager,
+    ) -> Self {
+        Self {
+            metadata: ObjectMeta {
+                name: Some(name.to_string()),
+                namespace: Some(namespace.to_string()),
+                uid: Some(uid.to_string()),
+                ..ObjectMeta::default()
+            },
+            name,
+            namespace,
+            image,
+            kafka_security,
+            authorization_config,
+            role_groups,
+            pod_descriptors,
+            metadata_manager,
+        }
+    }
+}
+
+/// Lets [`ValidatedKafkaCluster`] act as the owner [`Resource`] for child objects, so owner
+/// references are built from it (via the captured `metadata`) rather than the raw CR.
+impl Resource for ValidatedKafkaCluster {
+    type DynamicType = <v1alpha1::KafkaCluster as Resource>::DynamicType;
+    type Scope = <v1alpha1::KafkaCluster as Resource>::Scope;
+
+    fn kind(dt: &Self::DynamicType) -> Cow<'_, str> {
+        v1alpha1::KafkaCluster::kind(dt)
+    }
+
+    fn group(dt: &Self::DynamicType) -> Cow<'_, str> {
+        v1alpha1::KafkaCluster::group(dt)
+    }
+
+    fn version(dt: &Self::DynamicType) -> Cow<'_, str> {
+        v1alpha1::KafkaCluster::version(dt)
+    }
+
+    fn plural(dt: &Self::DynamicType) -> Cow<'_, str> {
+        v1alpha1::KafkaCluster::plural(dt)
+    }
+
+    fn meta(&self) -> &ObjectMeta {
+        &self.metadata
+    }
+
+    fn meta_mut(&mut self) -> &mut ObjectMeta {
+        &mut self.metadata
+    }
+}
+
 pub struct ValidatedRoleGroupConfig {
     pub merged_config: AnyConfig,
     // DESIGN DECISION: overrides are resolved into flat maps HERE rather than stored
@@ -265,7 +341,7 @@ pub async fn reconcile_kafka(
         APP_NAME,
         OPERATOR_NAME,
         KAFKA_CONTROLLER_NAME,
-        &kafka.object_ref(&()),
+        &validated_cluster.object_ref(&()),
         ClusterResourceApplyStrategy::from(&kafka.spec.cluster_operation),
         &kafka.spec.object_overrides,
     )
@@ -306,16 +382,19 @@ pub async fn reconcile_kafka(
             let rolegroup_ref = kafka.rolegroup_ref(kafka_role, rolegroup_name);
 
             let rg_headless_service = build_rolegroup_headless_service(
-                kafka,
+                &validated_cluster,
                 &validated_cluster.image,
                 &rolegroup_ref,
                 &validated_cluster.kafka_security,
             )
             .context(BuildServiceSnafu)?;
 
-            let rg_metrics_service =
-                build_rolegroup_metrics_service(kafka, &validated_cluster.image, &rolegroup_ref)
-                    .context(BuildServiceSnafu)?;
+            let rg_metrics_service = build_rolegroup_metrics_service(
+                &validated_cluster,
+                &validated_cluster.image,
+                &rolegroup_ref,
+            )
+            .context(BuildServiceSnafu)?;
 
             let kafka_listeners = get_kafka_listener_config(
                 kafka,
@@ -416,8 +495,9 @@ pub async fn reconcile_kafka(
         }
     }
 
-    let discovery_cm = build_discovery_configmap(kafka, validated_cluster, &bootstrap_listeners)
-        .context(BuildDiscoveryConfigSnafu)?;
+    let discovery_cm =
+        build::discovery::build_discovery_configmap(&validated_cluster, &bootstrap_listeners)
+            .context(BuildDiscoveryConfigSnafu)?;
 
     cluster_resources
         .add(client, discovery_cm)
diff --git a/rust/operator-binary/src/controller/build/config_map.rs b/rust/operator-binary/src/controller/build/config_map.rs
@@ -121,12 +121,12 @@ pub fn build_rolegroup_config_map(
     cm_builder
         .metadata(
             ObjectMetaBuilder::new()
-                .name_and_namespace(kafka)
+                .name_and_namespace(validated_cluster)
                 .name(rolegroup.object_name())
-                .ownerreference_from_resource(kafka, None, Some(true))
+                .ownerreference_from_resource(validated_cluster, None, Some(true))
                 .context(ObjectMissingMetadataForOwnerRefSnafu)?
                 .with_recommended_labels(&build_recommended_labels(
-                    kafka,
+                    validated_cluster,
                     KAFKA_CONTROLLER_NAME,
                     &resolved_product_image.app_version_label_value,
                     &rolegroup.role,
diff --git a/rust/operator-binary/src/controller/build/discovery.rs b/rust/operator-binary/src/controller/build/discovery.rs
@@ -5,7 +5,7 @@ use stackable_operator::{
     builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
     crd::listener,
     k8s_openapi::api::core::v1::ConfigMap,
-    kube::{ResourceExt, runtime::reflector::ObjectRef},
+    kube::runtime::reflector::ObjectRef,
 };
 
 use crate::{
@@ -22,9 +22,6 @@ pub enum Error {
         kafka: ObjectRef<v1alpha1::KafkaCluster>,
     },
 
-    #[snafu(display("object has no name associated"))]
-    NoName,
-
     #[snafu(display("could not find service port with name {}", port_name))]
     NoServicePort { port_name: String },
 
@@ -45,8 +42,7 @@ pub enum Error {
 /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain
 /// [`v1alpha1::KafkaCluster`].
 pub fn build_discovery_configmap(
-    owner: &v1alpha1::KafkaCluster,
-    validated_cluster: ValidatedKafkaCluster,
+    validated_cluster: &ValidatedKafkaCluster,
     listeners: &[listener::v1alpha1::Listener],
 ) -> Result<ConfigMap, Error> {
     let kafka_security = &validated_cluster.kafka_security;
@@ -68,14 +64,14 @@ pub fn build_discovery_configmap(
     ConfigMapBuilder::new()
         .metadata(
             ObjectMetaBuilder::new()
-                .name_and_namespace(owner)
-                .name(owner.name_unchecked())
-                .ownerreference_from_resource(owner, None, Some(true))
+                .name_and_namespace(validated_cluster)
+                .name(validated_cluster.name.to_string())
+                .ownerreference_from_resource(validated_cluster, None, Some(true))
                 .with_context(|_| ObjectMissingMetadataForOwnerRefSnafu {
-                    kafka: ObjectRef::from_obj(owner),
+                    kafka: cluster_object_ref(validated_cluster),
                 })?
                 .with_recommended_labels(&build_recommended_labels(
-                    owner,
+                    validated_cluster,
                     KAFKA_CONTROLLER_NAME,
                     &resolved_product_image.product_version,
                     &KafkaRole::Broker.to_string(),
@@ -89,6 +85,12 @@ pub fn build_discovery_configmap(
         .context(BuildConfigMapSnafu)
 }
 
+/// An [`ObjectRef`] to the owning cluster, built from the validated identity — used only for
+/// error context.
+fn cluster_object_ref(cluster: &ValidatedKafkaCluster) -> ObjectRef<v1alpha1::KafkaCluster> {
+    ObjectRef::new(cluster.name.as_ref()).within(cluster.namespace.as_ref())
+}
+
 fn listener_hosts(
     listeners: &[listener::v1alpha1::Listener],
     port_name: &str,
diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs
@@ -1,4 +1,5 @@
 //! Builders that assemble Kubernetes resources for kafka rolegroups.
 
 pub mod config_map;
+pub mod discovery;
 pub mod properties;
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
@@ -3,14 +3,21 @@
 //! Synchronously validates inputs that don't require a Kubernetes client. Produces
 //! [`ValidatedKafkaCluster`], consumed by the rest of `reconcile_kafka`.
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, str::FromStr};
 
-use snafu::{ResultExt, Snafu};
+use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     cli::OperatorEnvironmentOptions,
     commons::product_image_selection,
     config::merge::{Merge, merge},
-    v2::config_overrides::KeyValueConfigOverrides,
+    kube::ResourceExt,
+    v2::{
+        config_overrides::KeyValueConfigOverrides,
+        types::{
+            kubernetes::{NamespaceName, Uid},
+            operator::ClusterName,
+        },
+    },
 };
 
 use crate::{
@@ -50,6 +57,27 @@ pub enum Error {
 
     #[snafu(display("invalid metadata manager"))]
     InvalidMetadataManager { source: crate::crd::Error },
+
+    #[snafu(display("invalid cluster name"))]
+    InvalidClusterName {
+        source: stackable_operator::v2::macros::attributed_string_type::Error,
+    },
+
+    #[snafu(display("object defines no namespace"))]
+    ObjectHasNoNamespace,
+
+    #[snafu(display("invalid cluster namespace"))]
+    InvalidNamespace {
+        source: stackable_operator::v2::macros::attributed_string_type::Error,
+    },
+
+    #[snafu(display("object has no uid"))]
+    ObjectHasNoUid,
+
+    #[snafu(display("invalid cluster uid"))]
+    InvalidUid {
+        source: stackable_operator::v2::macros::attributed_string_type::Error,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -164,14 +192,22 @@ pub fn validate(
         .effective_metadata_manager()
         .context(InvalidMetadataManagerSnafu)?;
 
-    Ok(ValidatedKafkaCluster {
+    let name = ClusterName::from_str(&kafka.name_any()).context(InvalidClusterNameSnafu)?;
+    let namespace = NamespaceName::from_str(&kafka.namespace().context(ObjectHasNoNamespaceSnafu)?)
+        .context(InvalidNamespaceSnafu)?;
+    let uid = Uid::from_str(&kafka.uid().context(ObjectHasNoUidSnafu)?).context(InvalidUidSnafu)?;
+
+    Ok(ValidatedKafkaCluster::new(
+        name,
+        namespace,
+        uid,
         image,
         kafka_security,
-        authorization_config: dereferenced_objects.authorization_config,
+        dereferenced_objects.authorization_config,
         role_groups,
         pod_descriptors,
         metadata_manager,
-    })
+    ))
 }
 
 /// Merge role-group overrides over the role-level overrides (role-group wins per key) via the
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
@@ -43,7 +43,6 @@ use crate::{
 mod config;
 mod controller;
 mod crd;
-mod discovery;
 mod kerberos;
 mod operations;
 mod product_logging;
diff --git a/rust/operator-binary/src/resource/listener.rs b/rust/operator-binary/src/resource/listener.rs
@@ -36,12 +36,12 @@ pub fn build_broker_rolegroup_bootstrap_listener(
 
     Ok(listener::v1alpha1::Listener {
         metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(kafka)
+            .name_and_namespace(validated_cluster)
             .name(kafka.bootstrap_service_name(rolegroup))
-            .ownerreference_from_resource(kafka, None, Some(true))
+            .ownerreference_from_resource(validated_cluster, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
             .with_recommended_labels(&build_recommended_labels(
-                kafka,
+                validated_cluster,
                 KAFKA_CONTROLLER_NAME,
                 &resolved_product_image.app_version_label_value,
                 &rolegroup.role,
diff --git a/rust/operator-binary/src/resource/service.rs b/rust/operator-binary/src/resource/service.rs
diff --git a/rust/operator-binary/src/resource/statefulset.rs b/rust/operator-binary/src/resource/statefulset.rs
diff --git a/rust/operator-binary/src/utils.rs b/rust/operator-binary/src/utils.rs
