add headless and metrics services

Author: maltesander

rust/operator-binary/src/kafka_controller.rs

@@ -49,7 +49,7 @@ use crate::{
     resource::{
         configmap::build_rolegroup_config_map,
         listener::build_broker_rolegroup_bootstrap_listener,
-        service::build_rolegroup_service,
+        service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
         statefulset::{build_broker_rolegroup_statefulset, build_controller_rolegroup_statefulset},
     },
 };
@@ -347,8 +347,16 @@ pub async fn reconcile_kafka(
                 .merged_config(kafka, &rolegroup_ref.role_group)
                 .context(FailedToResolveConfigSnafu)?;
 
-            let rg_service =
-                build_rolegroup_service(kafka, &resolved_product_image, &rolegroup_ref)
+            let rg_headless_service = build_rolegroup_headless_service(
+                kafka,
+                &resolved_product_image,
+                &rolegroup_ref,
+                &kafka_security,
+            )
+            .context(BuildServiceSnafu)?;
+
+            let rg_metrics_service =
+                build_rolegroup_metrics_service(kafka, &resolved_product_image, &rolegroup_ref)
                     .context(BuildServiceSnafu)?;
 
             let rg_configmap = build_rolegroup_config_map(
@@ -407,7 +415,13 @@ pub async fn reconcile_kafka(
             }
 
             cluster_resources
-                .add(client, rg_service)
+                .add(client, rg_headless_service)
+                .await
+                .with_context(|_| ApplyRoleGroupServiceSnafu {
+                    rolegroup: rolegroup_ref.clone(),
+                })?;
+            cluster_resources
+                .add(client, rg_metrics_service)
                 .await
                 .with_context(|_| ApplyRoleGroupServiceSnafu {
                     rolegroup: rolegroup_ref.clone(),

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

@@ -2,13 +2,13 @@ use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
     commons::product_image_selection::ResolvedProductImage,
-    k8s_openapi::api::core::v1::{Service, ServiceSpec},
-    kvp::{Label, Labels},
+    k8s_openapi::api::core::v1::{Service, ServicePort, ServiceSpec},
+    kvp::{Annotations, Labels},
     role_utils::RoleGroupRef,
 };
 
 use crate::{
-    crd::{APP_NAME, v1alpha1},
+    crd::{APP_NAME, METRICS_PORT, METRICS_PORT_NAME, security::KafkaTlsSecurity, v1alpha1},
     kafka_controller::KAFKA_CONTROLLER_NAME,
     utils::build_recommended_labels,
 };
@@ -34,15 +34,16 @@ pub enum Error {
 /// 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_service(
+pub fn build_rolegroup_headless_service(
     kafka: &v1alpha1::KafkaCluster,
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
+    kafka_security: &KafkaTlsSecurity,
 ) -> Result<Service, Error> {
     Ok(Service {
         metadata: ObjectMetaBuilder::new()
             .name_and_namespace(kafka)
-            .name(rolegroup.object_name())
+            .name(rolegroup_headless_service_name(rolegroup))
             .ownerreference_from_resource(kafka, None, Some(true))
             .context(ObjectMissingMetadataForOwnerRefSnafu)?
             .with_recommended_labels(build_recommended_labels(
@@ -53,10 +54,10 @@ pub fn build_rolegroup_service(
                 &rolegroup.role_group,
             ))
             .context(MetadataBuildSnafu)?
-            .with_label(Label::try_from(("prometheus.io/scrape", "true")).context(LabelBuildSnafu)?)
             .build(),
         spec: Some(ServiceSpec {
             cluster_ip: Some("None".to_string()),
+            ports: Some(headless_ports(kafka_security)),
             selector: Some(
                 Labels::role_group_selector(
                     kafka,
@@ -73,3 +74,100 @@ pub fn build_rolegroup_service(
         status: None,
     })
 }
+
+/// The rolegroup metrics [`Service`] is a service that exposes metrics and a prometheus scraping label
+pub fn build_rolegroup_metrics_service(
+    kafka: &v1alpha1::KafkaCluster,
+    resolved_product_image: &ResolvedProductImage,
+    rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
+) -> Result<Service, Error> {
+    let metrics_service = Service {
+        metadata: ObjectMetaBuilder::new()
+            .name_and_namespace(kafka)
+            // TODO: Use method on RoleGroupRef once op-rs is released
+            .name(rolegroup_metrics_service_name(rolegroup))
+            .ownerreference_from_resource(kafka, None, Some(true))
+            .context(ObjectMissingMetadataForOwnerRefSnafu)?
+            .with_recommended_labels(build_recommended_labels(
+                kafka,
+                KAFKA_CONTROLLER_NAME,
+                &resolved_product_image.app_version_label_value,
+                &rolegroup.role,
+                &rolegroup.role_group,
+            ))
+            .context(MetadataBuildSnafu)?
+            .with_labels(prometheus_labels())
+            .with_annotations(prometheus_annotations())
+            .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(
+                    kafka,
+                    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
+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()),
+        port: METRICS_PORT.into(),
+        protocol: Some("TCP".to_string()),
+        ..ServicePort::default()
+    }]
+}
+
+fn headless_ports(kafka_security: &KafkaTlsSecurity) -> Vec<ServicePort> {
+    vec![ServicePort {
+        name: Some(kafka_security.client_port_name().into()),
+        port: kafka_security.client_port().into(),
+        protocol: Some("TCP".to_string()),
+        ..ServicePort::default()
+    }]
+}
+
+/// Common labels for Prometheus
+fn prometheus_labels() -> Labels {
+    Labels::try_from([("prometheus.io/scrape", "true")]).expect("should be a valid label")
+}
+
+/// Common annotations for Prometheus
+///
+/// These annotations can be used in a ServiceMonitor.
+///
+/// see also <https://github.com/prometheus-community/helm-charts/blob/prometheus-27.32.0/charts/prometheus/values.yaml#L983-L1036>
+fn prometheus_annotations() -> Annotations {
+    Annotations::try_from([
+        ("prometheus.io/path".to_owned(), "/metrics".to_owned()),
+        ("prometheus.io/port".to_owned(), METRICS_PORT.to_string()),
+        ("prometheus.io/scheme".to_owned(), "http".to_owned()),
+        ("prometheus.io/scrape".to_owned(), "true".to_owned()),
+    ])
+    .expect("should be valid annotations")
+}