Moving services to own module, splitting up functions

Maleware · Maleware · commit 9f2a57ffdaf8 · 2025-07-03T18:48:50.000+02:00
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
@@ -43,8 +43,8 @@ use stackable_operator::{
         api::{
             apps::v1::{StatefulSet, StatefulSetSpec},
             core::v1::{
-                ConfigMap, ConfigMapVolumeSource, EmptyDirVolumeSource, Probe, Service,
-                ServiceSpec, TCPSocketAction, Volume,
+                ConfigMap, ConfigMapVolumeSource, EmptyDirVolumeSource, Probe, TCPSocketAction,
+                Volume,
             },
         },
         apimachinery::pkg::{
@@ -56,7 +56,7 @@ use stackable_operator::{
         core::{DeserializeGuard, error_boundary},
         runtime::controller::Action,
     },
-    kvp::{Label, Labels, ObjectLabels},
+    kvp::{Labels, ObjectLabels},
     logging::controller::ReconcilerError,
     memory::{BinaryMultiple, MemoryQuantity},
     product_config_utils::{transform_all_roles_to_config, validate_all_roles_and_groups_config},
@@ -102,6 +102,10 @@ use crate::{
     listener::{LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_role_listener},
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::extend_role_group_config_map,
+    service::{
+        build_rolegroup_headless_service, build_rolegroup_metrics_service,
+        rolegroup_metrics_service_name,
+    },
 };
 
 pub const HIVE_CONTROLLER_NAME: &str = "hivecluster";
@@ -345,6 +349,9 @@ pub enum Error {
     BuildListenerVolume {
         source: ListenerOperatorVolumeSourceBuilderError,
     },
+
+    #[snafu(display("faild to configure service"))]
+    ServiceConfiguration { source: crate::service::Error },
 }
 type Result<T, E = Error> = std::result::Result<T, E>;
 
@@ -456,7 +463,14 @@ pub async fn reconcile_hive(
             .merged_config(&HiveRole::MetaStore, &rolegroup)
             .context(FailedToResolveResourceConfigSnafu)?;
 
-        let rg_services = build_rolegroup_services(hive, &resolved_product_image, &rolegroup)?;
+        let rg_metrics_service =
+            build_rolegroup_metrics_service(hive, &resolved_product_image, &rolegroup)
+                .context(ServiceConfigurationSnafu)?;
+
+        let rg_headless_service =
+            build_rolegroup_headless_service(hive, &resolved_product_image, &rolegroup)
+                .context(ServiceConfigurationSnafu)?;
+
         let rg_configmap = build_metastore_rolegroup_config_map(
             hive,
             &hive_namespace,
@@ -478,13 +492,19 @@ pub async fn reconcile_hive(
             &rbac_sa.name_any(),
         )?;
 
-        for rg_service in rg_services {
-            cluster_resources.add(client, rg_service).await.context(
-                ApplyRoleGroupServiceSnafu {
-                    rolegroup: rolegroup.clone(),
-                },
-            )?;
-        }
+        cluster_resources
+            .add(client, rg_metrics_service)
+            .await
+            .context(ApplyRoleGroupServiceSnafu {
+                rolegroup: rolegroup.clone(),
+            })?;
+
+        cluster_resources
+            .add(client, rg_headless_service)
+            .await
+            .context(ApplyRoleGroupServiceSnafu {
+                rolegroup: rolegroup.clone(),
+            })?;
 
         cluster_resources
             .add(client, rg_configmap)
@@ -714,97 +734,10 @@ 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_services(
-    hive: &v1alpha1::HiveCluster,
-    resolved_product_image: &ResolvedProductImage,
-    rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
-) -> 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,
-        },
-        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_services`]).
+/// corresponding [`Service`] (from [`build_rolegroup_headless_service`] and [`build_rolegroup_metrics_service`]).
 #[allow(clippy::too_many_arguments)]
 fn build_metastore_rolegroup_statefulset(
     hive: &v1alpha1::HiveCluster,
@@ -1147,7 +1080,7 @@ fn build_metastore_rolegroup_statefulset(
                 ..LabelSelector::default()
             },
             // TODO: Use method on RoleGroupRef once op-rs is released
-            service_name: Some(hive.rolegroup_headless_metrics_service_name(rolegroup_ref)),
+            service_name: Some(rolegroup_metrics_service_name(rolegroup_ref)),
             template: pod_template,
             volume_claim_templates: Some(vec![pvc]),
             ..StatefulSetSpec::default()
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
@@ -18,7 +18,7 @@ use stackable_operator::{
         merge::Merge,
     },
     crd::s3,
-    k8s_openapi::{api::core::v1::ServicePort, apimachinery::pkg::api::resource::Quantity},
+    k8s_openapi::apimachinery::pkg::api::resource::Quantity,
     kube::{CustomResource, ResourceExt, runtime::reflector::ObjectRef},
     product_config_utils::{self, Configuration},
     product_logging::{self, spec::Logging},
@@ -246,43 +246,6 @@ 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>,
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
@@ -7,6 +7,7 @@ mod kerberos;
 mod listener;
 mod operations;
 mod product_logging;
+mod service;
 
 use std::sync::Arc;
 
diff --git a/rust/operator-binary/src/service.rs b/rust/operator-binary/src/service.rs
@@ -0,0 +1,141 @@
+use snafu::{ResultExt, Snafu};
+use stackable_operator::{
+    builder::meta::ObjectMetaBuilder,
+    commons::product_image_selection::ResolvedProductImage,
+    k8s_openapi::api::core::v1::{Service, ServicePort, ServiceSpec},
+    kvp::{Label, Labels},
+    role_utils::RoleGroupRef,
+};
+
+use crate::{
+    controller::build_recommended_labels,
+    crd::{APP_NAME, HIVE_PORT, HIVE_PORT_NAME, METRICS_PORT, METRICS_PORT_NAME, v1alpha1},
+};
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("object is missing metadata to build owner reference"))]
+    ObjectMissingMetadataForOwnerRef {
+        source: stackable_operator::builder::meta::Error,
+    },
+    #[snafu(display("failed to build Metadata"))]
+    MetadataBuild {
+        source: stackable_operator::builder::meta::Error,
+    },
+    #[snafu(display("failed to build Labels"))]
+    LabelBuild {
+        source: stackable_operator::kvp::LabelError,
+    },
+}
+
+/// 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.
+pub fn build_rolegroup_headless_service(
+    hive: &v1alpha1::HiveCluster,
+    resolved_product_image: &ResolvedProductImage,
+    rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
+) -> Result<Service, Error> {
+    let headless_service = Service {
+        metadata: ObjectMetaBuilder::new()
+            .name_and_namespace(hive)
+            // TODO: Use method on RoleGroupRef once op-rs is released
+            .name(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(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(headless_service)
+}
+
+/// The rolegroup metrics [`Service`] is a service that exposes metrics and a prometheus scraping label
+pub fn build_rolegroup_metrics_service(
+    hive: &v1alpha1::HiveCluster,
+    resolved_product_image: &ResolvedProductImage,
+    rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
+) -> Result<Service, Error> {
+    let metrics_service = Service {
+        metadata: ObjectMetaBuilder::new()
+            .name_and_namespace(hive)
+            // TODO: Use method on RoleGroupRef once op-rs is released
+            .name(rolegroup_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(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,
+    };
+    Ok(metrics_service)
+}
+
+/// Headless service for cluster internal purposes only.
+// TODO: Move to operator-rs
+pub fn rolegroup_headless_service_name(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_metrics_service_name(rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>) -> String {
+    format!("{name}-metrics", name = rolegroup.object_name())
+}
+
+fn metrics_ports() -> Vec<ServicePort> {
+    vec![ServicePort {
+        name: Some(METRICS_PORT_NAME.to_string()),
+        port: METRICS_PORT.into(),
+        protocol: Some("TCP".to_string()),
+        ..ServicePort::default()
+    }]
+}
+
+fn service_ports() -> Vec<ServicePort> {
+    vec![ServicePort {
+        name: Some(HIVE_PORT_NAME.to_string()),
+        port: HIVE_PORT.into(),
+        protocol: Some("TCP".to_string()),
+        ..ServicePort::default()
+    }]
+}
