Consolidate listener to module

Author: Maleware

rust/operator-binary/src/controller.rs

@@ -37,10 +37,7 @@ use stackable_operator::{
         product_image_selection::ResolvedProductImage, rbac::build_rbac_resources,
         tls_verification::TlsClientDetailsError,
     },
-    crd::{
-        listener::v1alpha1::{Listener, ListenerPort, ListenerSpec},
-        s3,
-    },
+    crd::{listener::v1alpha1::Listener, s3},
     k8s_openapi::{
         DeepMerge,
         api::{
@@ -91,9 +88,8 @@ 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,
-        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,
+        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::{self, HiveMetastoreRoleConfig},
@@ -103,6 +99,7 @@ use crate::{
         self, add_kerberos_pod_config, kerberos_config_properties,
         kerberos_container_start_commands,
     },
+    listener::{LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_group_listener},
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::extend_role_group_config_map,
 };
@@ -341,6 +338,9 @@ pub enum Error {
         source: stackable_operator::cluster_resources::Error,
         role: String,
     },
+    #[snafu(display("failed to configure listener"))]
+    ListenerConfiguration { source: crate::listener::Error },
+
     #[snafu(display("failed to build listener volume"))]
     BuildListenerVolume {
         source: ListenerOperatorVolumeSourceBuilderError,
@@ -521,7 +521,8 @@ pub async fn reconcile_hive(
 
     if let Some(HiveMetastoreRoleConfig { listener_class, .. }) = role_config {
         let group_listener: Listener =
-            build_group_listener(hive, &resolved_product_image, &hive_role, listener_class)?;
+            build_group_listener(hive, &resolved_product_image, &hive_role, listener_class)
+                .context(ListenerConfigurationSnafu)?;
         let listener = cluster_resources
             .add(client, group_listener)
             .await
@@ -573,51 +574,6 @@ pub async fn reconcile_hive(
     Ok(Action::await_change())
 }
 
-// Designed to build a listener per role
-// In case of Hive we expect only one role: Metastore
-pub fn build_group_listener(
-    hive: &v1alpha1::HiveCluster,
-    resolved_product_image: &ResolvedProductImage,
-    hive_role: &HiveRole,
-    listener_class: &String,
-) -> Result<Listener> {
-    let metadata = ObjectMetaBuilder::new()
-        .name_and_namespace(hive)
-        .name(hive.group_listener_name(hive_role))
-        .ownerreference_from_resource(hive, None, Some(true))
-        .context(ObjectMissingMetadataForOwnerRefSnafu)?
-        .with_recommended_labels(build_recommended_labels(
-            hive,
-            &resolved_product_image.app_version_label,
-            &hive_role.to_string(),
-            "none",
-        ))
-        .context(MetadataBuildSnafu)?
-        .build();
-
-    let spec = ListenerSpec {
-        class_name: Some(listener_class.to_owned()),
-        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
 #[allow(clippy::too_many_arguments)]
 fn build_metastore_rolegroup_config_map(

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

@@ -32,7 +32,7 @@ use stackable_operator::{
 use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
 use v1alpha1::HiveMetastoreRoleConfig;
 
-use crate::crd::affinity::get_affinity;
+use crate::{crd::affinity::get_affinity, listener::metastore_default_listener_class};
 
 pub mod affinity;
 pub mod security;
@@ -61,13 +61,6 @@ pub const HIVE_PORT: u16 = 9083;
 pub const METRICS_PORT_NAME: &str = "metrics";
 pub const METRICS_PORT: u16 = 9084;
 
-// Listener volumes
-pub const LISTENER_VOLUME_NAME: &str = "listener";
-pub const LISTENER_VOLUME_DIR: &str = "/stackable/listener";
-
-// Listener defaults
-pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal";
-
 // Certificates and trust stores
 pub const SYSTEM_TRUST_STORE: &str = "/etc/pki/java/cacerts";
 pub const SYSTEM_TRUST_STORE_PASSWORD: &str = "changeit";
@@ -190,10 +183,6 @@ impl Default for v1alpha1::HiveMetastoreRoleConfig {
     }
 }
 
-fn metastore_default_listener_class() -> String {
-    DEFAULT_LISTENER_CLASS.to_owned()
-}
-
 impl HasStatusCondition for v1alpha1::HiveCluster {
     fn conditions(&self) -> Vec<ClusterCondition> {
         match &self.status {

rust/operator-binary/src/discovery.rs

@@ -9,7 +9,8 @@ use stackable_operator::{
 
 use crate::{
     controller::build_recommended_labels,
-    crd::{HIVE_PORT_NAME, HiveRole, v1alpha1},
+    crd::{HiveRole, v1alpha1},
+    listener::build_listener_connection_string,
 };
 
 #[derive(Snafu, Debug)]
@@ -19,25 +20,21 @@ pub enum Error {
         source: stackable_operator::builder::meta::Error,
         hive: ObjectRef<v1alpha1::HiveCluster>,
     },
-    #[snafu(display("chroot path {chroot} was relative (must be absolute)"))]
-    RelativeChroot { chroot: String },
+
     #[snafu(display("could not build discovery config map for {obj_ref}"))]
     DiscoveryConfigMap {
         source: stackable_operator::builder::configmap::Error,
         obj_ref: ObjectRef<v1alpha1::HiveCluster>,
     },
-    #[snafu(display("could not find port [{port_name}] for rolegroup listener {role}"))]
-    NoServicePort { port_name: String, role: String },
-
     #[snafu(display("invalid owner name for discovery ConfigMap"))]
     InvalidOwnerNameForDiscoveryConfigMap,
 
     #[snafu(display("failed to build Metadata"))]
     MetadataBuild {
         source: stackable_operator::builder::meta::Error,
     },
-    #[snafu(display("{role} listener has no adress"))]
-    RoleListenerHasNoAddress { role: String },
+    #[snafu(display("failed to configure listener discovery configmap"))]
+    ListenerConfiguration { source: crate::listener::Error },
 }
 
 /// Builds discovery [`ConfigMap`]s for connecting to a [`v1alpha1::HiveCluster`] for all expected
@@ -104,7 +101,8 @@ fn build_discovery_configmap(
 
     discovery_configmap.add_data(
         "HIVE".to_string(),
-        build_listener_connection_string(listener, &hive_role.to_string(), chroot)?,
+        build_listener_connection_string(listener, &hive_role.to_string(), chroot)
+            .context(ListenerConfigurationSnafu)?,
     );
 
     discovery_configmap
@@ -114,38 +112,6 @@ fn build_discovery_configmap(
         })
 }
 
-// Builds the connection string with respect to the listener provided objects
-fn build_listener_connection_string(
-    listener_ref: Listener,
-    role: &String,
-    chroot: Option<&str>,
-) -> Result<String, Error> {
-    // We only need the first address corresponding to the role
-    let listener_address = listener_ref
-        .status
-        .and_then(|s| s.ingress_addresses?.into_iter().next())
-        .context(RoleListenerHasNoAddressSnafu { role })?;
-    let mut conn_str = format!(
-        "thrift://{address}:{port}",
-        address = listener_address.address,
-        port = listener_address
-            .ports
-            .get(HIVE_PORT_NAME)
-            .copied()
-            .context(NoServicePortSnafu {
-                port_name: HIVE_PORT_NAME,
-                role
-            })?
-    );
-    if let Some(chroot) = chroot {
-        if !chroot.starts_with('/') {
-            return RelativeChrootSnafu { chroot }.fail();
-        }
-        conn_str.push_str(chroot);
-    }
-    Ok(conn_str)
-}
-
 pub fn build_headless_role_group_metrics_service_name(name: String) -> String {
     format!("{name}-metrics", name = name)
 }

rust/operator-binary/src/listener.rs

@@ -0,0 +1,118 @@
+use snafu::{OptionExt, ResultExt, Snafu};
+use stackable_operator::{
+    builder::meta::ObjectMetaBuilder,
+    commons::product_image_selection::ResolvedProductImage,
+    crd::listener::v1alpha1::{Listener, ListenerPort, ListenerSpec},
+};
+
+use crate::{
+    controller::build_recommended_labels,
+    crd::{HIVE_PORT, HIVE_PORT_NAME, HiveRole, v1alpha1},
+};
+
+// Listener volumes
+pub const LISTENER_VOLUME_NAME: &str = "listener";
+pub const LISTENER_VOLUME_DIR: &str = "/stackable/listener";
+
+// Listener defaults
+pub const DEFAULT_LISTENER_CLASS: &str = "cluster-internal";
+
+#[derive(Snafu, Debug)]
+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("{role} listener has no adress"))]
+    RoleListenerHasNoAddress { role: String },
+    #[snafu(display("could not find port [{port_name}] for rolegroup listener {role}"))]
+    NoServicePort { port_name: String, role: String },
+    #[snafu(display("chroot path {chroot} was relative (must be absolute)"))]
+    RelativeChroot { chroot: String },
+}
+
+// Builds the connection string with respect to the listener provided objects
+pub fn build_listener_connection_string(
+    listener_ref: Listener,
+    role: &String,
+    chroot: Option<&str>,
+) -> Result<String, Error> {
+    // We only need the first address corresponding to the role
+    let listener_address = listener_ref
+        .status
+        .and_then(|s| s.ingress_addresses?.into_iter().next())
+        .context(RoleListenerHasNoAddressSnafu { role })?;
+    let mut conn_str = format!(
+        "thrift://{address}:{port}",
+        address = listener_address.address,
+        port = listener_address
+            .ports
+            .get(HIVE_PORT_NAME)
+            .copied()
+            .context(NoServicePortSnafu {
+                port_name: HIVE_PORT_NAME,
+                role
+            })?
+    );
+    if let Some(chroot) = chroot {
+        if !chroot.starts_with('/') {
+            return RelativeChrootSnafu { chroot }.fail();
+        }
+        conn_str.push_str(chroot);
+    }
+    Ok(conn_str)
+}
+
+// Designed to build a listener per role
+// In case of Hive we expect only one role: Metastore
+pub fn build_group_listener(
+    hive: &v1alpha1::HiveCluster,
+    resolved_product_image: &ResolvedProductImage,
+    hive_role: &HiveRole,
+    listener_class: &String,
+) -> Result<Listener, Error> {
+    let metadata = ObjectMetaBuilder::new()
+        .name_and_namespace(hive)
+        .name(hive.group_listener_name(hive_role))
+        .ownerreference_from_resource(hive, None, Some(true))
+        .context(ObjectMissingMetadataForOwnerRefSnafu)?
+        .with_recommended_labels(build_recommended_labels(
+            hive,
+            &resolved_product_image.app_version_label,
+            &hive_role.to_string(),
+            "none",
+        ))
+        .context(MetadataBuildSnafu)?
+        .build();
+
+    let spec = ListenerSpec {
+        class_name: Some(listener_class.to_owned()),
+        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()),
+    }]
+}
+
+// used by crds to define a default listener_class name
+pub fn metastore_default_listener_class() -> String {
+    DEFAULT_LISTENER_CLASS.to_owned()
+}

rust/operator-binary/src/main.rs

@@ -4,6 +4,7 @@ mod controller;
 mod crd;
 mod discovery;
 mod kerberos;
+mod listener;
 mod operations;
 mod product_logging;