add headleass & metrics service

Author: maltesander

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

@@ -187,7 +187,7 @@ pub fn controller_kafka_container_command(
 fn to_listeners(port: u16) -> String {
     // The environment variables are set in the statefulset of the controller
     format!(
-        "{listener_name}://$POD_NAME.$ROLEGROUP_REF.$NAMESPACE.svc.$CLUSTER_DOMAIN:{port}",
+        "{listener_name}://$POD_NAME.$ROLEGROUP_HEADLESS_SERVICE_NAME.$NAMESPACE.svc.$CLUSTER_DOMAIN:{port}",
         listener_name = KafkaListenerName::Controller
     )
 }

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

@@ -294,7 +294,9 @@ impl v1alpha1::KafkaCluster {
                     for replica in 0..replicas {
                         pod_descriptors.push(KafkaPodDescriptor {
                             namespace: namespace.clone(),
-                            role_group_service_name: rolegroup_ref.object_name(),
+                            role_group_service_name: rolegroup_ref
+                                .rolegroup_headless_service_name(),
+                            role_group_statefulset_name: rolegroup_ref.object_name(),
                             replica,
                             cluster_domain: cluster_info.cluster_domain.clone(),
                             node_id: node_id_hash_offset + u32::from(replica),
@@ -341,6 +343,7 @@ impl v1alpha1::KafkaCluster {
 #[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub struct KafkaPodDescriptor {
     namespace: String,
+    role_group_statefulset_name: String,
     role_group_service_name: String,
     replica: u16,
     cluster_domain: DomainName,
@@ -361,7 +364,7 @@ impl KafkaPodDescriptor {
     }
 
     pub fn pod_name(&self) -> String {
-        format!("{}-{}", self.role_group_service_name, self.replica)
+        format!("{}-{}", self.role_group_statefulset_name, self.replica)
     }
 
     /// Build the Kraft voter String

rust/operator-binary/src/resource/service.rs

@@ -43,7 +43,7 @@ pub fn build_rolegroup_headless_service(
     Ok(Service {
         metadata: ObjectMetaBuilder::new()
             .name_and_namespace(kafka)
-            .name(rolegroup_headless_service_name(rolegroup))
+            .name(rolegroup.rolegroup_headless_service_name())
             .ownerreference_from_resource(kafka, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
             .with_recommended_labels(build_recommended_labels(
@@ -85,7 +85,7 @@ pub fn build_rolegroup_metrics_service(
         metadata: ObjectMetaBuilder::new()
             .name_and_namespace(kafka)
             // TODO: Use method on RoleGroupRef once op-rs is released
-            .name(rolegroup_metrics_service_name(rolegroup))
+            .name(rolegroup.rolegroup_metrics_service_name())
             .ownerreference_from_resource(kafka, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
             .with_recommended_labels(build_recommended_labels(
@@ -122,18 +122,6 @@ pub fn build_rolegroup_metrics_service(
     Ok(metrics_service)
 }
 
-/// Headless service for cluster internal purposes only.
-// TODO: Move to operator-rs
-fn rolegroup_headless_service_name(rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>) -> String {
-    format!("{name}-headless", name = rolegroup.object_name())
-}
-
-/// Headless metrics service exposes Prometheus endpoint only
-// TODO: Move to operator-rs
-fn rolegroup_metrics_service_name(rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>) -> String {
-    format!("{name}-metrics", name = rolegroup.object_name())
-}
-
 fn metrics_ports() -> Vec<ServicePort> {
     vec![ServicePort {
         name: Some(METRICS_PORT_NAME.to_string()),

rust/operator-binary/src/resource/statefulset.rs

@@ -547,7 +547,7 @@ pub fn build_broker_rolegroup_statefulset(
                 ),
                 ..LabelSelector::default()
             },
-            service_name: Some(rolegroup_ref.object_name()),
+            service_name: Some(rolegroup_ref.rolegroup_headless_service_name()),
             template: pod_template,
             volume_claim_templates: Some(pvcs),
             ..StatefulSetSpec::default()
@@ -621,8 +621,8 @@ pub fn build_controller_rolegroup_statefulset(
     });
 
     env.push(EnvVar {
-        name: "ROLEGROUP_REF".to_string(),
-        value: Some(rolegroup_ref.object_name()),
+        name: "ROLEGROUP_HEADLESS_SERVICE_NAME".to_string(),
+        value: Some(rolegroup_ref.rolegroup_headless_service_name()),
         ..EnvVar::default()
     });
 
@@ -875,7 +875,7 @@ pub fn build_controller_rolegroup_statefulset(
                 ),
                 ..LabelSelector::default()
             },
-            service_name: Some(rolegroup_ref.object_name()),
+            service_name: Some(rolegroup_ref.rolegroup_headless_service_name()),
             template: pod_template,
             volume_claim_templates: Some(merged_config.resources().storage.build_pvcs()),
             ..StatefulSetSpec::default()