create separate headless services

Author: labrenbe

rust/operator-binary/src/controller.rs

@@ -37,8 +37,8 @@ use stackable_operator::{
             apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy},
             core::v1::{
                 ConfigMap, ConfigMapKeySelector, ConfigMapVolumeSource, EmptyDirVolumeSource,
-                EnvVar, EnvVarSource, ObjectFieldSelector, Probe, SecretVolumeSource, Service,
-                ServicePort, ServiceSpec, TCPSocketAction, Volume,
+                EnvVar, EnvVarSource, ObjectFieldSelector, Probe, SecretVolumeSource,
+                TCPSocketAction, Volume,
             },
         },
         apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
@@ -48,7 +48,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::env_vars_from_rolegroup_config,
@@ -107,6 +107,10 @@ use crate::{
         build_tls_volume, check_or_generate_oidc_admin_password, check_or_generate_sensitive_key,
         tls::{KEYSTORE_NIFI_CONTAINER_MOUNT, KEYSTORE_VOLUME_NAME, TRUSTSTORE_VOLUME_NAME},
     },
+    service::{
+        build_rolegroup_headless_service, build_rolegroup_metrics_service,
+        rolegroup_headless_service_name,
+    },
 };
 
 pub const NIFI_CONTROLLER_NAME: &str = "nificluster";
@@ -346,6 +350,9 @@ pub enum Error {
 
     #[snafu(display("failed to configure listener"))]
     ListenerConfiguration { source: crate::listener::Error },
+
+    #[snafu(display("failed to configure service"))]
+    ServiceConfiguration { source: crate::service::Error },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -488,8 +495,24 @@ pub async fn reconcile_nifi(
             )
             .context(InvalidGitSyncSpecSnafu)?;
 
-            let rg_service =
-                build_node_rolegroup_service(nifi, &resolved_product_image, &rolegroup)?;
+            let role_group_service_recommended_labels = build_recommended_labels(
+                nifi,
+                &resolved_product_image.app_version_label,
+                &rolegroup.role,
+                &rolegroup.role_group,
+            );
+
+            let role_group_service_selector =
+                Labels::role_group_selector(nifi, APP_NAME, &rolegroup.role, &rolegroup.role_group)
+                    .context(LabelBuildSnafu)?;
+
+            let rg_headless_service = build_rolegroup_headless_service(
+                nifi,
+                &rolegroup,
+                role_group_service_recommended_labels.clone(),
+                role_group_service_selector.clone().into(),
+            )
+            .context(ServiceConfigurationSnafu)?;
 
             let role = nifi.spec.nodes.as_ref().context(NoNodesDefinedSnafu)?;
 
@@ -539,12 +562,30 @@ pub async fn reconcile_nifi(
             )
             .await?;
 
+            if resolved_product_image.product_version.starts_with("1.") {
+                let rg_metrics_service = build_rolegroup_metrics_service(
+                    nifi,
+                    &rolegroup,
+                    role_group_service_recommended_labels,
+                    role_group_service_selector.into(),
+                )
+                .context(ServiceConfigurationSnafu)?;
+
+                cluster_resources
+                    .add(client, rg_metrics_service)
+                    .await
+                    .with_context(|_| ApplyRoleGroupServiceSnafu {
+                        rolegroup: rolegroup.clone(),
+                    })?;
+            }
+
             cluster_resources
-                .add(client, rg_service)
+                .add(client, rg_headless_service)
                 .await
                 .with_context(|_| ApplyRoleGroupServiceSnafu {
                     rolegroup: rolegroup.clone(),
                 })?;
+
             cluster_resources
                 .add(client, rg_configmap)
                 .await
@@ -778,76 +819,12 @@ async fn build_node_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_node_rolegroup_service(
-    nifi: &v1alpha1::NifiCluster,
-    resolved_product_image: &ResolvedProductImage,
-    rolegroup: &RoleGroupRef<v1alpha1::NifiCluster>,
-) -> Result<Service> {
-    let metadata = ObjectMetaBuilder::new()
-        .name_and_namespace(nifi)
-        .name(rolegroup_service_name(rolegroup))
-        .ownerreference_from_resource(nifi, None, Some(true))
-        .context(ObjectMissingMetadataForOwnerRefSnafu)?
-        .with_recommended_labels(build_recommended_labels(
-            nifi,
-            &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();
-
-    // In NiFi 2.x metrics are scraped from the HTTPS port
-    let mut service_ports = vec![];
-    if resolved_product_image.product_version.starts_with("1.") {
-        service_ports.push(ServicePort {
-            name: Some(METRICS_PORT_NAME.to_owned()),
-            port: METRICS_PORT.into(),
-            protocol: Some("TCP".to_owned()),
-            ..ServicePort::default()
-        });
-    } else {
-        service_ports.push(ServicePort {
-            name: Some(HTTPS_PORT_NAME.to_owned()),
-            port: HTTPS_PORT.into(),
-            protocol: Some("TCP".to_owned()),
-            ..ServicePort::default()
-        });
-    }
-
-    let spec = Some(ServiceSpec {
-        // Internal communication does not need to be exposed
-        type_: Some("ClusterIP".to_owned()),
-        cluster_ip: Some("None".to_owned()),
-        ports: Some(service_ports),
-        selector: Some(
-            Labels::role_group_selector(nifi, APP_NAME, &rolegroup.role, &rolegroup.role_group)
-                .context(LabelBuildSnafu)?
-                .into(),
-        ),
-        publish_not_ready_addresses: Some(true),
-        ..ServiceSpec::default()
-    });
-
-    let service = Service {
-        metadata,
-        spec,
-        status: None,
-    };
-
-    Ok(service)
-}
-
 const USERDATA_MOUNTPOINT: &str = "/stackable/userdata";
 
 /// 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_node_rolegroup_service`]).
+/// corresponding [`stackable_operator::k8s_openapi::api::core::v1::Service`] (from [`build_rolegroup_headless_service`]).
 #[allow(clippy::too_many_arguments)]
 async fn build_node_rolegroup_statefulset(
     nifi: &v1alpha1::NifiCluster,
@@ -939,7 +916,7 @@ async fn build_node_rolegroup_statefulset(
 
     let node_address = format!(
         "$POD_NAME.{service_name}.{namespace}.svc.{cluster_domain}",
-        service_name = rolegroup_service_name(rolegroup_ref),
+        service_name = rolegroup_headless_service_name(&rolegroup_ref.object_name()),
         namespace = &nifi
             .metadata
             .namespace
@@ -1443,7 +1420,9 @@ async fn build_node_rolegroup_statefulset(
                 ),
                 ..LabelSelector::default()
             },
-            service_name: Some(rolegroup_service_name(rolegroup_ref)),
+            service_name: Some(rolegroup_headless_service_name(
+                &rolegroup_ref.object_name(),
+            )),
             template: pod_template,
             update_strategy: Some(StatefulSetUpdateStrategy {
                 type_: if rolling_update_supported {
@@ -1482,10 +1461,6 @@ async fn build_node_rolegroup_statefulset(
     })
 }
 
-pub fn rolegroup_service_name(rolegroup: &RoleGroupRef<v1alpha1::NifiCluster>) -> String {
-    format!("{name}-headless", name = rolegroup.object_name())
-}
-
 async fn get_proxy_hosts(
     client: &Client,
     nifi: &v1alpha1::NifiCluster,

rust/operator-binary/src/main.rs

@@ -39,6 +39,7 @@ mod operations;
 mod product_logging;
 mod reporting_task;
 mod security;
+mod service;
 
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));