WIP: Listener integration

Author: Maleware

Cargo.lock

@@ -77,23 +77,6 @@ spec:
                       required:
                         - configMap
                       type: object
-                    listenerClass:
-                      default: cluster-internal
-                      description: |-
-                        This field controls which type of Service the Operator creates for this HiveCluster:
-
-                        * cluster-internal: Use a ClusterIP service
-
-                        * external-unstable: Use a NodePort service
-
-                        * external-stable: Use a LoadBalancer service
-
-                        This is a temporary solution with the goal to keep yaml manifests forward compatible. In the future, this setting will control which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) will be used to expose the service, and ListenerClass names will stay the same, allowing for a non-breaking change.
-                      enum:
-                        - cluster-internal
-                        - external-unstable
-                        - external-stable
-                      type: string
                     s3:
                       description: S3 connection specification. This can be either `inline` or a `reference` to an S3Connection object. Read the [S3 concept documentation](https://docs.stackable.tech/home/nightly/concepts/s3) to learn more.
                       nullable: true
@@ -333,6 +316,10 @@ spec:
                           description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                           nullable: true
                           type: string
+                        listenerClass:
+                          description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+                          nullable: true
+                          type: string
                         logging:
                           default:
                             containers: {}
@@ -627,6 +614,10 @@ spec:
                                 description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
                                 nullable: true
                                 type: string
+                              listenerClass:
+                                description: This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver.
+                                nullable: true
+                                type: string
                               logging:
                                 default:
                                   containers: {}

deploy/helm/hive-operator/crds/crds.yaml

@@ -3,6 +3,7 @@
 use std::{
     borrow::Cow,
     collections::{BTreeMap, HashMap},
+    default,
     hash::Hasher,
     sync::Arc,
 };
@@ -22,16 +23,28 @@ use stackable_operator::{
         configmap::ConfigMapBuilder,
         meta::ObjectMetaBuilder,
         pod::{
-            PodBuilder, container::ContainerBuilder, resources::ResourceRequirementsBuilder,
-            security::PodSecurityContextBuilder, volume::VolumeBuilder,
+            PodBuilder,
+            container::ContainerBuilder,
+            resources::ResourceRequirementsBuilder,
+            security::PodSecurityContextBuilder,
+            volume::{
+                ListenerOperatorVolumeSourceBuilder, ListenerOperatorVolumeSourceBuilderError,
+                ListenerReference, VolumeBuilder,
+            },
         },
     },
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
     commons::{
         product_image_selection::ResolvedProductImage, rbac::build_rbac_resources,
         tls_verification::TlsClientDetailsError,
     },
-    crd::s3,
+    crd::{
+        listener::{
+            self,
+            v1alpha1::{Listener, ListenerPort, ListenerSpec},
+        },
+        s3,
+    },
     k8s_openapi::{
         DeepMerge,
         api::{
@@ -82,8 +95,9 @@ use crate::{
     crd::{
         APP_NAME, CORE_SITE_XML, Container, DB_PASSWORD_ENV, DB_USERNAME_ENV, HIVE_PORT,
         HIVE_PORT_NAME, HIVE_SITE_XML, HiveClusterStatus, HiveRole, JVM_SECURITY_PROPERTIES_FILE,
-        METRICS_PORT, METRICS_PORT_NAME, MetaStoreConfig, STACKABLE_CONFIG_DIR,
-        STACKABLE_CONFIG_DIR_NAME, STACKABLE_CONFIG_MOUNT_DIR, STACKABLE_CONFIG_MOUNT_DIR_NAME,
+        LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, METRICS_PORT, METRICS_PORT_NAME,
+        MetaStoreConfig, STACKABLE_CONFIG_DIR, STACKABLE_CONFIG_DIR_NAME,
+        STACKABLE_CONFIG_MOUNT_DIR, STACKABLE_CONFIG_MOUNT_DIR_NAME,
         STACKABLE_LOG_CONFIG_MOUNT_DIR, STACKABLE_LOG_CONFIG_MOUNT_DIR_NAME, STACKABLE_LOG_DIR,
         STACKABLE_LOG_DIR_NAME, v1alpha1,
     },
@@ -327,6 +341,16 @@ pub enum Error {
 
     #[snafu(display("failed to construct JVM arguments"))]
     ConstructJvmArguments { source: crate::config::jvm::Error },
+
+    #[snafu(display("failed to apply group listener for {rolegroup}"))]
+    ApplyGroupListener {
+        source: stackable_operator::cluster_resources::Error,
+        rolegroup: RoleGroupRef<v1alpha1::HiveCluster>,
+    },
+    #[snafu(display("failed to build listener volume"))]
+    BuildListenerVolume {
+        source: ListenerOperatorVolumeSourceBuilderError,
+    },
 }
 type Result<T, E = Error> = std::result::Result<T, E>;
 
@@ -428,16 +452,10 @@ pub async fn reconcile_hive(
         .await
         .context(ApplyRoleBindingSnafu)?;
 
-    let metastore_role_service = build_metastore_role_service(hive, &resolved_product_image)?;
-
-    // we have to get the assigned ports
-    let metastore_role_service = cluster_resources
-        .add(client, metastore_role_service)
-        .await
-        .context(ApplyRoleServiceSnafu)?;
-
     let mut ss_cond_builder = StatefulSetConditionBuilder::default();
 
+    // TODO: You want to add service creation in this for loop
+    // for each rg a service needs to be build for hive
     for (rolegroup_name, rolegroup_config) in metastore_config.iter() {
         let rolegroup = hive.metastore_rolegroup_ref(rolegroup_name);
 
@@ -467,6 +485,20 @@ pub async fn reconcile_hive(
             &rbac_sa.name_any(),
         )?;
 
+        let rg_group_listener = build_group_listener(
+            hive,
+            &resolved_product_image,
+            &rolegroup,
+            config.listener_class,
+        )?;
+
+        cluster_resources
+            .add(client, rg_group_listener)
+            .await
+            .with_context(|_| ApplyGroupListenerSnafu {
+                rolegroup: rolegroup.clone(),
+            })?;
+
         cluster_resources
             .add(client, rg_service)
             .await
@@ -504,16 +536,10 @@ pub async fn reconcile_hive(
     // std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases.
     // We don't /need/ stability, but it's still nice to avoid spurious changes where possible.
     let mut discovery_hash = FnvHasher::with_key(0);
-    for discovery_cm in discovery::build_discovery_configmaps(
-        client,
-        hive,
-        hive,
-        &resolved_product_image,
-        &metastore_role_service,
-        None,
-    )
-    .await
-    .context(BuildDiscoveryConfigSnafu)?
+    for discovery_cm in
+        discovery::build_discovery_configmaps(client, hive, hive, &resolved_product_image, None)
+            .await
+            .context(BuildDiscoveryConfigSnafu)?
     {
         let discovery_cm = cluster_resources
             .add(client, discovery_cm)
@@ -547,43 +573,47 @@ pub async fn reconcile_hive(
     Ok(Action::await_change())
 }
 
-/// The server-role service is the primary endpoint that should be used by clients that do not
-/// perform internal load balancing including targets outside of the cluster.
-pub fn build_metastore_role_service(
+pub fn build_group_listener(
     hive: &v1alpha1::HiveCluster,
     resolved_product_image: &ResolvedProductImage,
-) -> Result<Service> {
-    let role_name = HiveRole::MetaStore.to_string();
+    rolegroup: &RoleGroupRef<v1alpha1::HiveCluster>,
+    listener_class: String,
+) -> Result<Listener> {
+    let metadata = ObjectMetaBuilder::new()
+        .name_and_namespace(hive)
+        .name(hive.group_listener_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();
 
-    let role_svc_name = hive
-        .metastore_role_service_name()
-        .context(GlobalServiceNameNotFoundSnafu)?;
-    Ok(Service {
-        metadata: ObjectMetaBuilder::new()
-            .name_and_namespace(hive)
-            .name(role_svc_name)
-            .ownerreference_from_resource(hive, None, Some(true))
-            .context(ObjectMissingMetadataForOwnerRefSnafu)?
-            .with_recommended_labels(build_recommended_labels(
-                hive,
-                &resolved_product_image.app_version_label,
-                &role_name,
-                "global",
-            ))
-            .context(MetadataBuildSnafu)?
-            .build(),
-        spec: Some(ServiceSpec {
-            type_: Some(hive.spec.cluster_config.listener_class.k8s_service_type()),
-            ports: Some(service_ports()),
-            selector: Some(
-                Labels::role_selector(hive, APP_NAME, &role_name)
-                    .context(LabelBuildSnafu)?
-                    .into(),
-            ),
-            ..ServiceSpec::default()
-        }),
+    let spec = ListenerSpec {
+        class_name: Some(listener_class),
+        ports: Some(listener_ports()),
+        ..Default::default()
+    };
+
+    let listener = Listener {
+        metadata,
+        spec,
         status: None,
-    })
+    };
+
+    Ok(listener)
+}
+
+fn listener_ports() -> Vec<ListenerPort> {
+    vec![ListenerPort {
+        name: HIVE_PORT_NAME.to_owned(),
+        port: HIVE_PORT.into(),
+        protocol: Some("TCP".to_owned()),
+    }]
 }
 
 /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
@@ -729,6 +759,7 @@ 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.
+// TODO: However it looks like I need to add it here as services will be created here by rolegroup
 fn build_rolegroup_service(
     hive: &v1alpha1::HiveCluster,
     resolved_product_image: &ResolvedProductImage,
@@ -949,16 +980,39 @@ fn build_metastore_rolegroup_statefulset(
         }
     }
 
+    let recommended_object_labels = build_recommended_labels(
+        hive,
+        &resolved_product_image.app_version_label,
+        &rolegroup_ref.role,
+        &rolegroup_ref.role_group,
+    );
+    // Used for PVC templates that cannot be modified once they are deployed
+    let unversioned_recommended_labels = Labels::recommended(build_recommended_labels(
+        hive,
+        // A version value is required, and we do want to use the "recommended" format for the other desired labels
+        "none",
+        &rolegroup_ref.role,
+        &rolegroup_ref.role_group,
+    ))
+    .context(LabelBuildSnafu)?;
+
     let metadata = ObjectMetaBuilder::new()
-        .with_recommended_labels(build_recommended_labels(
-            hive,
-            &resolved_product_image.app_version_label,
-            &rolegroup_ref.role,
-            &rolegroup_ref.role_group,
-        ))
+        .with_recommended_labels(recommended_object_labels.clone())
         .context(MetadataBuildSnafu)?
         .build();
 
+    let pvc = ListenerOperatorVolumeSourceBuilder::new(
+        &ListenerReference::ListenerName(hive.group_listener_name(rolegroup_ref)),
+        &unversioned_recommended_labels,
+    )
+    .context(BuildListenerVolumeSnafu)?
+    .build_pvc(LISTENER_VOLUME_NAME.to_owned())
+    .context(BuildListenerVolumeSnafu)?;
+
+    container_builder
+        .add_volume_mount(LISTENER_VOLUME_NAME, LISTENER_VOLUME_DIR)
+        .context(AddVolumeMountSnafu)?;
+
     pod_builder
         .metadata(metadata)
         .image_pull_secrets_from_product_image(resolved_product_image)
@@ -1069,12 +1123,7 @@ fn build_metastore_rolegroup_statefulset(
             .name(rolegroup_ref.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_ref.role,
-                &rolegroup_ref.role_group,
-            ))
+            .with_recommended_labels(recommended_object_labels)
             .context(MetadataBuildSnafu)?
             .build(),
         spec: Some(StatefulSetSpec {
@@ -1095,6 +1144,7 @@ fn build_metastore_rolegroup_statefulset(
             },
             service_name: Some(rolegroup_ref.object_name()),
             template: pod_template,
+            volume_claim_templates: Some(vec![pvc]),
             ..StatefulSetSpec::default()
         }),
         status: None,