Listener: Use headless and metrics service

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
   - Use `--file-log-rotation-period` (or `FILE_LOG_ROTATION_PERIOD`) to configure the frequency of rotation.
   - Use `--console-log-format` (or `CONSOLE_LOG_FORMAT`) to set the format to `plain` (default) or `json`.
 - BREAKING: Add Listener support for Hive ([#605]).
+- Add internal headless service in addition to the metrics service ([#613]).
 
 ### Changed
 
@@ -43,6 +44,7 @@ All notable changes to this project will be documented in this file.
 [#603]: https://github.com/stackabletech/hive-operator/pull/603
 [#604]: https://github.com/stackabletech/hive-operator/pull/604
 [#605]: https://github.com/stackabletech/hive-operator/pull/605
+[#613]: https://github.com/stackabletech/hive-operator/pull/613
 
 ## [25.3.0] - 2025-03-21
 

rust/operator-binary/src/controller.rs

@@ -44,7 +44,7 @@ use stackable_operator::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
                 ConfigMap, ConfigMapVolumeSource, EmptyDirVolumeSource, Probe, Service,
-                ServicePort, ServiceSpec, TCPSocketAction, Volume,
+                ServiceSpec, TCPSocketAction, Volume,
             },
         },
         apimachinery::pkg::{
@@ -94,7 +94,7 @@ use crate::{
         STACKABLE_LOG_DIR_NAME,
         v1alpha1::{self, HiveMetastoreRoleConfig},
     },
-    discovery::{self, build_headless_role_group_metrics_service_name},
+    discovery::{self},
     kerberos::{
         self, add_kerberos_pod_config, kerberos_config_properties,
         kerberos_container_start_commands,
@@ -456,7 +456,7 @@ pub async fn reconcile_hive(
             .merged_config(&HiveRole::MetaStore, &rolegroup)
             .context(FailedToResolveResourceConfigSnafu)?;
 
-        let rg_service = build_rolegroup_service(hive, &resolved_product_image, &rolegroup)?;
+        let rg_services = build_rolegroup_services(hive, &resolved_product_image, &rolegroup)?;
         let rg_configmap = build_metastore_rolegroup_config_map(
             hive,
             &hive_namespace,
@@ -478,12 +478,13 @@ pub async fn reconcile_hive(
             &rbac_sa.name_any(),
         )?;
 
-        cluster_resources
-            .add(client, rg_service)
-            .await
-            .context(ApplyRoleGroupServiceSnafu {
-                rolegroup: rolegroup.clone(),
-            })?;
+        for rg_service in rg_services {
+            cluster_resources.add(client, rg_service).await.context(
+                ApplyRoleGroupServiceSnafu {
+                    rolegroup: rolegroup.clone(),
+                },
+            )?;
+        }
 
         cluster_resources
             .add(client, rg_configmap)
@@ -716,49 +717,94 @@ fn build_metastore_rolegroup_config_map(
 /// The rolegroup [`Service`] is a headless service that allows direct access to the instances of a certain rolegroup
 ///
 /// This is mostly useful for internal communication between peers, or for clients that perform client-side load balancing.
-fn build_rolegroup_service(
+fn build_rolegroup_services(
     hive: &v1alpha1::HiveCluster,
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
-) -> Result<Service> {
-    Ok(Service {
-        metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(hive)
-            .name(build_headless_role_group_metrics_service_name(
-                rolegroup.object_name(),
-            ))
-            .ownerreference_from_resource(hive, None, Some(true))
-            .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(build_recommended_labels(
-                hive,
-                &resolved_product_image.app_version_label,
-                &rolegroup.role,
-                &rolegroup.role_group,
-            ))
-            .context(MetadataBuildSnafu)?
-            .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
-            .build(),
-        spec: Some(ServiceSpec {
-            // Internal communication does not need to be exposed
-            type_: Some("ClusterIP".to_string()),
-            cluster_ip: Some("None".to_string()),
-            ports: Some(service_ports()),
-            selector: Some(
-                Labels::role_group_selector(hive, APP_NAME, &rolegroup.role, &rolegroup.role_group)
+) -> Result<Vec<Service>> {
+    let services = vec![
+        Service {
+            metadata: ObjectMetaBuilder::new()
+                .name_and_namespace(hive)
+                // TODO: Use method on RoleGroupRef once op-rs is released
+                .name(hive.rolegroup_headless_metrics_service_name(rolegroup))
+                .ownerreference_from_resource(hive, None, Some(true))
+                .context(ObjectMissingMetadataForOwnerRefSnafu)?
+                .with_recommended_labels(build_recommended_labels(
+                    hive,
+                    &resolved_product_image.app_version_label,
+                    &rolegroup.role,
+                    &rolegroup.role_group,
+                ))
+                .context(MetadataBuildSnafu)?
+                .with_label(
+                    Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?,
+                )
+                .build(),
+            spec: Some(ServiceSpec {
+                // Internal communication does not need to be exposed
+                type_: Some("ClusterIP".to_string()),
+                cluster_ip: Some("None".to_string()),
+                ports: Some(hive.metrics_ports()),
+                selector: Some(
+                    Labels::role_group_selector(
+                        hive,
+                        APP_NAME,
+                        &rolegroup.role,
+                        &rolegroup.role_group,
+                    )
                     .context(LabelBuildSnafu)?
                     .into(),
-            ),
-            publish_not_ready_addresses: Some(true),
-            ..ServiceSpec::default()
-        }),
-        status: None,
-    })
+                ),
+                publish_not_ready_addresses: Some(true),
+                ..ServiceSpec::default()
+            }),
+            status: None,
+        },
+        Service {
+            metadata: ObjectMetaBuilder::new()
+                .name_and_namespace(hive)
+                // TODO: Use method on RoleGroupRef once op-rs is released
+                .name(hive.rolegroup_headless_service_name(rolegroup))
+                .ownerreference_from_resource(hive, None, Some(true))
+                .context(ObjectMissingMetadataForOwnerRefSnafu)?
+                .with_recommended_labels(build_recommended_labels(
+                    hive,
+                    &resolved_product_image.app_version_label,
+                    &rolegroup.role,
+                    &rolegroup.role_group,
+                ))
+                .context(MetadataBuildSnafu)?
+                .build(),
+            spec: Some(ServiceSpec {
+                // Internal communication does not need to be exposed
+                type_: Some("ClusterIP".to_string()),
+                cluster_ip: Some("None".to_string()),
+                // Expecting same ports as on listener service, just as a headless, internal service
+                ports: Some(hive.service_ports()),
+                selector: Some(
+                    Labels::role_group_selector(
+                        hive,
+                        APP_NAME,
+                        &rolegroup.role,
+                        &rolegroup.role_group,
+                    )
+                    .context(LabelBuildSnafu)?
+                    .into(),
+                ),
+                publish_not_ready_addresses: Some(true),
+                ..ServiceSpec::default()
+            }),
+            status: None,
+        },
+    ];
+    Ok(services)
 }
 
 /// The rolegroup [`StatefulSet`] runs the rolegroup, as configured by the administrator.
 ///
 /// The [`Pod`](`stackable_operator::k8s_openapi::api::core::v1::Pod`)s are accessible through the
-/// corresponding [`Service`] (from [`build_rolegroup_service`]).
+/// corresponding [`Service`] (from [`build_rolegroup_services`]).
 #[allow(clippy::too_many_arguments)]
 fn build_metastore_rolegroup_statefulset(
     hive: &v1alpha1::HiveCluster,
@@ -1100,9 +1146,8 @@ fn build_metastore_rolegroup_statefulset(
                 ),
                 ..LabelSelector::default()
             },
-            service_name: Some(build_headless_role_group_metrics_service_name(
-                rolegroup_ref.object_name(),
-            )),
+            // TODO: Use method on RoleGroupRef once op-rs is released
+            service_name: Some(hive.rolegroup_headless_metrics_service_name(rolegroup_ref)),
             template: pod_template,
             volume_claim_templates: Some(vec![pvc]),
             ..StatefulSetSpec::default()
@@ -1123,15 +1168,6 @@ pub fn error_policy(
     }
 }
 
-pub fn service_ports() -> Vec<ServicePort> {
-    vec![ServicePort {
-        name: Some(METRICS_PORT_NAME.to_string()),
-        port: METRICS_PORT.into(),
-        protocol: Some("TCP".to_string()),
-        ..ServicePort::default()
-    }]
-}
-
 /// Creates recommended `ObjectLabels` to be used in deployed resources
 pub fn build_recommended_labels<'a, T>(
     owner: &'a T,

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

@@ -18,7 +18,7 @@ use stackable_operator::{
         merge::Merge,
     },
     crd::s3,
-    k8s_openapi::apimachinery::pkg::api::resource::Quantity,
+    k8s_openapi::{api::core::v1::ServicePort, apimachinery::pkg::api::resource::Quantity},
     kube::{CustomResource, ResourceExt, runtime::reflector::ObjectRef},
     product_config_utils::{self, Configuration},
     product_logging::{self, spec::Logging},
@@ -246,6 +246,43 @@ impl v1alpha1::HiveCluster {
         format!("{name}-{role}", name = self.name_any(), role = hive_role)
     }
 
+    /// Set of functions to define service names on rolegroup level.
+    /// Headless service for cluster internal purposes only.
+    // TODO: Move to operator-rs
+    pub fn rolegroup_headless_service_name(
+        &self,
+        rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
+    ) -> String {
+        format!("{name}-headless", name = rolegroup.object_name())
+    }
+
+    /// Headless metrics service exposes Prometheus endpoint only
+    // TODO: Move to operator-rs
+    pub fn rolegroup_headless_metrics_service_name(
+        &self,
+        rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
+    ) -> String {
+        format!("{name}-metrics", name = rolegroup.object_name())
+    }
+
+    pub fn metrics_ports(&self) -> Vec<ServicePort> {
+        vec![ServicePort {
+            name: Some(METRICS_PORT_NAME.to_string()),
+            port: METRICS_PORT.into(),
+            protocol: Some("TCP".to_string()),
+            ..ServicePort::default()
+        }]
+    }
+
+    pub fn service_ports(&self) -> Vec<ServicePort> {
+        vec![ServicePort {
+            name: Some(HIVE_PORT_NAME.to_string()),
+            port: HIVE_PORT.into(),
+            protocol: Some("TCP".to_string()),
+            ..ServicePort::default()
+        }]
+    }
+
     pub fn rolegroup(
         &self,
         rolegroup_ref: &RoleGroupRef<Self>,

rust/operator-binary/src/discovery.rs

@@ -111,7 +111,3 @@ fn build_discovery_configmap(
             obj_ref: ObjectRef::from_obj(hive),
         })
 }
-
-pub fn build_headless_role_group_metrics_service_name(name: String) -> String {
-    format!("{name}-metrics")
-}

rust/operator-binary/src/listener.rs

@@ -104,7 +104,7 @@ pub fn build_role_listener(
     Ok(listener)
 }
 
-fn listener_ports() -> Vec<ListenerPort> {
+pub fn listener_ports() -> Vec<ListenerPort> {
     vec![ListenerPort {
         name: HIVE_PORT_NAME.to_owned(),
         port: HIVE_PORT.into(),

tests/templates/kuttl/external-access/20-assert.yaml

@@ -37,3 +37,10 @@ metadata:
   name: test-hive-metastore-default-metrics
 spec:
   type: ClusterIP # exposed metrics
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: test-hive-metastore-default-headless
+spec:
+  type: ClusterIP # exposed metrics

tests/templates/kuttl/smoke/80-assert.yaml

@@ -2,4 +2,4 @@
 apiVersion: kuttl.dev/v1beta1
 kind: TestAssert
 commands:
-  - script: kubectl exec -n "$NAMESPACE" test-metastore-0 -- python /tmp/test_metastore.py -m hive-metastore.$NAMESPACE.svc.cluster.local
+  - script: kubectl exec -n "$NAMESPACE" test-metastore-0 -- python /tmp/test_metastore.py -m hive-metastore-default-headless.$NAMESPACE.svc.cluster.local