refactor: use v2 owerref, labels, RoleGroupName, cleanup

Author: maltesander

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

@@ -99,7 +99,7 @@ mod tests {
         let rg = validated
             .role_group_configs
             .get(&KafkaRole::Broker)
-            .and_then(|groups| groups.get("default"))
+            .and_then(|groups| groups.get(&"default".parse().unwrap()))
             .expect("broker default role group should exist");
         (rg.config.clone(), rg.jvm_argument_overrides.clone())
     }

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

@@ -1,19 +1,13 @@
-use stackable_operator::role_utils::RoleGroupRef;
-
-use crate::crd::v1alpha1::KafkaCluster;
+use crate::crd::role::KafkaRole;
 
 /// The Kafka node.id needs to be unique across the Kafka cluster.
 /// This function generates an integer that is stable for a given role group
 /// regardless if broker or controllers.
 /// This integer is then added to the pod index to compute the final node.id
 /// The node.id is only set and used in Kraft mode.
 /// Warning: this is not safe from collisions.
-pub fn node_id_hash32_offset(rolegroup_ref: &RoleGroupRef<KafkaCluster>) -> u32 {
-    let hash = fnv_hash32(&format!(
-        "{role}-{rolegroup}",
-        role = rolegroup_ref.role,
-        rolegroup = rolegroup_ref.role_group
-    ));
+pub fn node_id_hash32_offset(role: &KafkaRole, role_group: &str) -> u32 {
+    let hash = fnv_hash32(&format!("{role}-{role_group}"));
     let range = hash & 0x0000FFFF;
     // Kafka uses signed integer
     range * 0x00007FFF

rust/operator-binary/src/controller/build/config_map.rs

@@ -3,65 +3,70 @@ use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
     k8s_openapi::api::core::v1::ConfigMap,
-    role_utils::RoleGroupRef,
-    v2::config_file_writer::{PropertiesWriterError, to_java_properties_string},
+    product_logging::framework::VECTOR_CONFIG_FILE,
+    v2::{
+        builder::meta::ownerreference_from_resource,
+        config_file_writer::{PropertiesWriterError, to_java_properties_string},
+    },
 };
 
 use crate::{
     controller::{
-        ValidatedCluster, ValidatedRoleGroupConfig,
+        RoleGroupName, ValidatedCluster, ValidatedRoleGroupConfig,
         build::properties::logging::role_group_config_map_data,
     },
     crd::{
         ConfigFileName, STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR,
         listener::{KafkaListenerConfig, node_address_cmd},
         role::AnyConfig,
-        v1alpha1,
     },
-    kafka_controller::{KAFKA_CONTROLLER_NAME, build_recommended_labels},
 };
 
 #[derive(Snafu, Debug)]
 pub enum Error {
-    #[snafu(display("failed to build ConfigMap for {}", rolegroup))]
+    #[snafu(display("failed to build ConfigMap for role group {role_group}"))]
     BuildRoleGroupConfig {
         source: stackable_operator::builder::configmap::Error,
-        rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
+        role_group: RoleGroupName,
     },
 
-    #[snafu(display("failed to serialize [{}] for {rolegroup}", ConfigFileName::Security))]
+    #[snafu(display(
+        "failed to serialize [{}] for role group {role_group}",
+        ConfigFileName::Security
+    ))]
     JvmSecurityProperties {
         source: PropertiesWriterError,
-        rolegroup: String,
-    },
-
-    #[snafu(display("failed to build Metadata"))]
-    MetadataBuild {
-        source: stackable_operator::builder::meta::Error,
-    },
-
-    #[snafu(display("object is missing metadata to build owner reference"))]
-    ObjectMissingMetadataForOwnerRef {
-        source: stackable_operator::builder::meta::Error,
+        role_group: RoleGroupName,
     },
 
-    #[snafu(display("failed to serialize config for {rolegroup}"))]
+    #[snafu(display("failed to serialize config for role group {role_group}"))]
     SerializeConfig {
         source: PropertiesWriterError,
-        rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
+        role_group: RoleGroupName,
+    },
+
+    #[snafu(display("failed to build pod descriptors"))]
+    BuildPodDescriptors {
+        source: crate::controller::PodDescriptorsError,
     },
 
     #[snafu(display("no Kraft controllers found to build"))]
     NoKraftControllersFound,
 }
 
-/// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
+/// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator.
+///
+/// `vector_config` is the Vector agent config built by the caller (where a `RoleGroupRef` is
+/// available); it is `None` when the Vector agent is disabled. Resource naming and labels use the
+/// role (derived from `validated_rg.config`) and the typed `role_group_name`.
 pub fn build_rolegroup_config_map(
     validated_cluster: &ValidatedCluster,
-    rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
+    role_group_name: &RoleGroupName,
     validated_rg: &ValidatedRoleGroupConfig,
     listener_config: &KafkaListenerConfig,
+    vector_config: Option<String>,
 ) -> Result<ConfigMap, Error> {
+    let role = validated_rg.config.kafka_role();
     let cluster_config = &validated_cluster.cluster_config;
     let kafka_security = &cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
@@ -72,20 +77,26 @@ pub fn build_rolegroup_config_map(
         .overrides
         .clone();
 
-    if cluster_config.is_kraft_mode() && cluster_config.pod_descriptors.is_empty() {
+    let pod_descriptors = validated_cluster
+        .pod_descriptors(None)
+        .context(BuildPodDescriptorsSnafu)?;
+
+    if cluster_config.is_kraft_mode() && pod_descriptors.is_empty() {
         return NoKraftControllersFoundSnafu.fail();
     }
 
     let kafka_config = match &validated_rg.config {
         AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build(
             cluster_config,
             listener_config,
+            &pod_descriptors,
             config_overrides,
         ),
         AnyConfig::Controller(_) => {
             crate::controller::build::properties::controller_properties::build(
                 cluster_config,
                 listener_config,
+                &pod_descriptors,
                 config_overrides,
             )
         }
@@ -101,32 +112,33 @@ pub fn build_rolegroup_config_map(
         .metadata(
             ObjectMetaBuilder::new()
                 .name_and_namespace(validated_cluster)
-                .name(rolegroup.object_name())
-                .ownerreference_from_resource(validated_cluster, None, Some(true))
-                .context(ObjectMissingMetadataForOwnerRefSnafu)?
-                .with_recommended_labels(&build_recommended_labels(
+                .name(
+                    validated_cluster
+                        .resource_names(&role, role_group_name)
+                        .role_group_config_map()
+                        .to_string(),
+                )
+                .ownerreference(ownerreference_from_resource(
                     validated_cluster,
-                    KAFKA_CONTROLLER_NAME,
-                    &resolved_product_image.app_version_label_value,
-                    &rolegroup.role,
-                    &rolegroup.role_group,
+                    None,
+                    Some(true),
                 ))
-                .context(MetadataBuildSnafu)?
+                .with_labels(validated_cluster.recommended_labels(&role, role_group_name))
                 .build(),
         )
         .add_data(
             kafka_config_file_name,
             to_java_properties_string(kafka_config.iter()).with_context(|_| {
                 SerializeConfigSnafu {
-                    rolegroup: rolegroup.clone(),
+                    role_group: role_group_name.clone(),
                 }
             })?,
         )
         .add_data(
             ConfigFileName::Security.to_string(),
             to_java_properties_string(jvm_sec_props.iter()).with_context(|_| {
                 JvmSecurityPropertiesSnafu {
-                    rolegroup: rolegroup.role_group.clone(),
+                    role_group: role_group_name.clone(),
                 }
             })?,
         )
@@ -139,7 +151,7 @@ pub fn build_rolegroup_config_map(
                     .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))),
             )
             .with_context(|_| JvmSecurityPropertiesSnafu {
-                rolegroup: rolegroup.role_group.clone(),
+                role_group: role_group_name.clone(),
             })?,
         )
         // This file contains the JAAS configuration for Kerberos authentication
@@ -156,7 +168,6 @@ pub fn build_rolegroup_config_map(
 
     let config_data = role_group_config_map_data(
         &resolved_product_image.product_version,
-        rolegroup,
         &validated_rg.config,
     );
     for (file_name, data) in config_data {
@@ -165,10 +176,14 @@ pub fn build_rolegroup_config_map(
         }
     }
 
+    if let Some(vector_config) = vector_config {
+        cm_builder.add_data(VECTOR_CONFIG_FILE, vector_config);
+    }
+
     cm_builder
         .build()
         .with_context(|_| BuildRoleGroupConfigSnafu {
-            rolegroup: rolegroup.clone(),
+            role_group: role_group_name.clone(),
         })
 }
 

rust/operator-binary/src/controller/build/discovery.rs

@@ -1,27 +1,20 @@
-use std::num::TryFromIntError;
+use std::{num::TryFromIntError, str::FromStr};
 
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
     crd::listener,
     k8s_openapi::api::core::v1::ConfigMap,
-    kube::runtime::reflector::ObjectRef,
+    v2::builder::meta::ownerreference_from_resource,
 };
 
 use crate::{
-    controller::ValidatedCluster,
-    crd::{role::KafkaRole, v1alpha1},
-    kafka_controller::{KAFKA_CONTROLLER_NAME, build_recommended_labels},
+    controller::{RoleGroupName, ValidatedCluster},
+    crd::role::KafkaRole,
 };
 
 #[derive(Snafu, Debug)]
 pub enum Error {
-    #[snafu(display("object {} is missing metadata to build owner reference", kafka))]
-    ObjectMissingMetadataForOwnerRef {
-        source: stackable_operator::builder::meta::Error,
-        kafka: ObjectRef<v1alpha1::KafkaCluster>,
-    },
-
     #[snafu(display("could not find service port with name {}", port_name))]
     NoServicePort { port_name: String },
 
@@ -32,11 +25,6 @@ pub enum Error {
     BuildConfigMap {
         source: stackable_operator::builder::configmap::Error,
     },
-
-    #[snafu(display("failed to build metadata"))]
-    MetadataBuild {
-        source: stackable_operator::builder::meta::Error,
-    },
 }
 
 /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain
@@ -46,7 +34,6 @@ pub fn build_discovery_configmap(
     listeners: &[listener::v1alpha1::Listener],
 ) -> Result<ConfigMap, Error> {
     let kafka_security = &validated_cluster.cluster_config.kafka_security;
-    let resolved_product_image = &validated_cluster.image;
 
     let port_name = if kafka_security.has_kerberos_enabled() {
         kafka_security.bootstrap_port_name()
@@ -65,31 +52,25 @@ pub fn build_discovery_configmap(
         .metadata(
             ObjectMetaBuilder::new()
                 .name_and_namespace(validated_cluster)
-                .ownerreference_from_resource(validated_cluster, None, Some(true))
-                .with_context(|_| ObjectMissingMetadataForOwnerRefSnafu {
-                    kafka: cluster_object_ref(validated_cluster),
-                })?
-                .with_recommended_labels(&build_recommended_labels(
+                .ownerreference(ownerreference_from_resource(
                     validated_cluster,
-                    KAFKA_CONTROLLER_NAME,
-                    &resolved_product_image.product_version,
-                    &KafkaRole::Broker.to_string(),
-                    "discovery",
+                    None,
+                    Some(true),
                 ))
-                .context(MetadataBuildSnafu)?
+                .with_labels(
+                    validated_cluster.recommended_labels(
+                        &KafkaRole::Broker,
+                        &RoleGroupName::from_str("discovery")
+                            .expect("'discovery' is a valid role group name"),
+                    ),
+                )
                 .build(),
         )
         .add_data("KAFKA", bootstrap_servers)
         .build()
         .context(BuildConfigMapSnafu)
 }
 
-/// An [`ObjectRef`] to the owning cluster, built from the validated identity — used only for
-/// error context.
-fn cluster_object_ref(cluster: &ValidatedCluster) -> ObjectRef<v1alpha1::KafkaCluster> {
-    ObjectRef::new(cluster.name.as_ref()).within(cluster.namespace.as_ref())
-}
-
 fn listener_hosts(
     listeners: &[listener::v1alpha1::Listener],
     port_name: &str,

rust/operator-binary/src/controller/build/properties/broker_properties.rs

@@ -4,6 +4,7 @@ use super::kraft_controllers;
 use crate::{
     controller::ValidatedClusterConfig,
     crd::{
+        KafkaPodDescriptor,
         listener::{KafkaListenerConfig, KafkaListenerName},
         role::{
             KAFKA_ADVERTISED_LISTENERS, KAFKA_BROKER_ID, KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS,
@@ -17,9 +18,10 @@ use crate::{
 pub fn build(
     cluster_config: &ValidatedClusterConfig,
     listener_config: &KafkaListenerConfig,
+    pod_descriptors: &[KafkaPodDescriptor],
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(&cluster_config.pod_descriptors);
+    let kraft_controllers = kraft_controllers(pod_descriptors);
 
     let mut result = BTreeMap::from([
         (

rust/operator-binary/src/controller/build/properties/controller_properties.rs

@@ -4,6 +4,7 @@ use super::kraft_controllers;
 use crate::{
     controller::ValidatedClusterConfig,
     crd::{
+        KafkaPodDescriptor,
         listener::{KafkaListenerConfig, KafkaListenerName},
         role::{
             KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP,
@@ -16,9 +17,10 @@ use crate::{
 pub fn build(
     cluster_config: &ValidatedClusterConfig,
     listener_config: &KafkaListenerConfig,
+    pod_descriptors: &[KafkaPodDescriptor],
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(&cluster_config.pod_descriptors).join(",");
+    let kraft_controllers = kraft_controllers(pod_descriptors).join(",");
 
     let mut result = BTreeMap::from([
         (