create separate headless services and fix discovery cm

Author: labrenbe

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

@@ -32,10 +32,7 @@ use stackable_operator::{
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
     time::Duration,
-    utils::{
-        COMMON_BASH_TRAP_FUNCTIONS, cluster_info::KubernetesClusterInfo,
-        crds::raw_object_list_schema,
-    },
+    utils::{COMMON_BASH_TRAP_FUNCTIONS, crds::raw_object_list_schema},
     versioned::versioned,
 };
 use strum::{Display, EnumDiscriminants, EnumIter, EnumString, IntoStaticStr};
@@ -470,25 +467,6 @@ impl v1alpha1::DruidCluster {
         .collect()
     }
 
-    /// The name of the role-level load-balanced Kubernetes `Service`
-    pub fn role_service_name(&self, role: &DruidRole) -> Option<String> {
-        Some(format!("{}-{}-headless", self.metadata.name.clone()?, role))
-    }
-
-    /// The fully-qualified domain name of the role-level load-balanced Kubernetes `Service`
-    pub fn role_service_fqdn(
-        &self,
-        role: &DruidRole,
-        cluster_info: &KubernetesClusterInfo,
-    ) -> Option<String> {
-        Some(format!(
-            "{service_name}.{namespace}.svc.{cluster_domain}",
-            service_name = self.role_service_name(role)?,
-            namespace = self.metadata.namespace.as_ref()?,
-            cluster_domain = cluster_info.cluster_domain,
-        ))
-    }
-
     /// If an s3 connection for ingestion is given, as well as an s3 connection for deep storage, they need to be the same.
     /// This function returns the resolved connection, or raises an Error if the connections are not identical.
     pub async fn get_s3_connection(
@@ -1665,32 +1643,6 @@ where
 
 #[cfg(test)]
 mod tests {
-    use stackable_operator::commons::networking::DomainName;
-
-    use super::*;
-
-    #[test]
-    fn test_service_name_generation() {
-        let cluster = deserialize_yaml_file::<v1alpha1::DruidCluster>(
-            "test/resources/crd/role_service/druid_cluster.yaml",
-        );
-        let dummy_cluster_info = KubernetesClusterInfo {
-            cluster_domain: DomainName::try_from("cluster.local").unwrap(),
-        };
-
-        assert_eq!(cluster.metadata.name, Some("testcluster".to_string()));
-
-        assert_eq!(
-            cluster.role_service_name(&DruidRole::Router),
-            Some("testcluster-router-headless".to_string())
-        );
-
-        assert_eq!(
-            cluster.role_service_fqdn(&DruidRole::Router, &dummy_cluster_info),
-            Some("testcluster-router-headless.default.svc.cluster.local".to_string())
-        )
-    }
-
     pub fn deserialize_yaml_str<'a, T: serde::de::Deserialize<'a>>(value: &'a str) -> T {
         let deserializer = serde_yaml::Deserializer::from_str(value);
         serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap()

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

@@ -15,7 +15,7 @@ use stackable_operator::{
     },
     crd::listener,
     k8s_openapi::{
-        api::core::v1::{ContainerPort, Probe, TCPSocketAction},
+        api::core::v1::{ContainerPort, Probe, ServicePort, TCPSocketAction},
         apimachinery::pkg::util::intstr::IntOrString,
     },
     time::Duration,
@@ -58,8 +58,8 @@ const PLAINTEXT_PORT: &str = "druid.plaintextPort";
 const ENABLE_TLS_PORT: &str = "druid.enableTlsPort";
 const TLS_PORT: &str = "druid.tlsPort";
 // Port names
-const PLAINTEXT_PORT_NAME: &str = "http";
-const TLS_PORT_NAME: &str = "https";
+pub const PLAINTEXT_PORT_NAME: &str = "http";
+pub const TLS_PORT_NAME: &str = "https";
 // Client side (Druid) TLS
 const CLIENT_HTTPS_KEY_STORE_PATH: &str = "druid.client.https.keyStorePath";
 const CLIENT_HTTPS_KEY_STORE_TYPE: &str = "druid.client.https.keyStoreType";
@@ -162,6 +162,18 @@ impl DruidTlsSecurity {
             .collect()
     }
 
+    pub fn service_ports(&self, role: &DruidRole) -> Vec<ServicePort> {
+        self.exposed_ports(role)
+            .into_iter()
+            .map(|(name, val)| ServicePort {
+                name: Some(name),
+                port: val.into(),
+                protocol: Some("TCP".to_string()),
+                ..ServicePort::default()
+            })
+            .collect()
+    }
+
     pub fn listener_ports(
         &self,
         role: &DruidRole,

rust/operator-binary/src/discovery.rs

@@ -1,18 +1,19 @@
 //! Discovery for Druid.  We make Druid discoverable by putting a connection string to the router service
 //! inside a config map.  We only provide a connection string to the router service, since it serves as
 //! a gateway to the cluster for client queries.
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
     commons::product_image_selection::ResolvedProductImage,
+    crd::listener::v1alpha1::Listener,
     k8s_openapi::api::core::v1::ConfigMap,
     kube::{Resource, ResourceExt, runtime::reflector::ObjectRef},
-    utils::cluster_info::KubernetesClusterInfo,
 };
 
 use crate::{
     DRUID_CONTROLLER_NAME,
     crd::{DruidRole, build_recommended_labels, security::DruidTlsSecurity, v1alpha1},
+    listener::build_listener_connection_string,
 };
 
 #[derive(Snafu, Debug)]
@@ -35,47 +36,45 @@ pub enum Error {
     AddRecommendedLabels {
         source: stackable_operator::builder::meta::Error,
     },
+
+    #[snafu(display("failed to configure listener discovery configmap"))]
+    ListenerConfiguration { source: crate::listener::Error },
 }
 
 /// Builds discovery [`ConfigMap`]s for connecting to a [`v1alpha1::DruidCluster`].
 pub async fn build_discovery_configmaps(
     druid: &v1alpha1::DruidCluster,
     owner: &impl Resource<DynamicType = ()>,
-    cluster_info: &KubernetesClusterInfo,
     resolved_product_image: &ResolvedProductImage,
     druid_tls_security: &DruidTlsSecurity,
+    listener: Listener,
 ) -> Result<Vec<ConfigMap>, Error> {
     let name = owner.name_unchecked();
     Ok(vec![build_discovery_configmap(
         druid,
         owner,
-        cluster_info,
         resolved_product_image,
         druid_tls_security,
         &name,
+        listener,
     )?])
 }
 
 /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain [`v1alpha1::DruidCluster`].
 fn build_discovery_configmap(
     druid: &v1alpha1::DruidCluster,
     owner: &impl Resource<DynamicType = ()>,
-    cluster_info: &KubernetesClusterInfo,
     resolved_product_image: &ResolvedProductImage,
     druid_tls_security: &DruidTlsSecurity,
     name: &str,
+    listener: Listener,
 ) -> Result<ConfigMap, Error> {
-    let router_host = format!(
-        "{}:{}",
-        druid
-            .role_service_fqdn(&DruidRole::Router, cluster_info)
-            .with_context(|| NoServiceFqdnSnafu)?,
-        if druid_tls_security.tls_enabled() {
-            DruidRole::Router.get_https_port()
-        } else {
-            DruidRole::Router.get_http_port()
-        }
-    );
+    let router_host = build_listener_connection_string(
+        listener,
+        druid_tls_security,
+        &DruidRole::Router.to_string(),
+    )
+    .context(ListenerConfigurationSnafu)?;
     let sqlalchemy_conn_str = format!("druid://{}/druid/v2/sql", router_host);
     let avatica_conn_str = format!(
         "jdbc:avatica:remote:url=http://{}/druid/v2/sql/avatica/",