resolve listener-class across role-groups/role

Author: adwk67

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

@@ -583,6 +583,12 @@ impl v1alpha1::HbaseCluster {
         }
     }
 
+    /// Returns rolegroup and replica information for a specific role.
+    /// We can't pass through the merged config for a particular role-group
+    /// here as we need more than the config. As this will be called by role,
+    /// the merged listener-class is called so that only role-group information
+    /// for externally-reachable services (based on their listener class) are
+    /// included in the collection.
     pub fn rolegroup_ref_and_replicas(
         &self,
         role: &HbaseRole,
@@ -596,6 +602,9 @@ impl v1alpha1::HbaseCluster {
                 // Order rolegroups consistently, to avoid spurious downstream rewrites
                 .collect::<BTreeMap<_, _>>()
                 .into_iter()
+                .filter(|(rolegroup_name, _)| {
+                    self.resolved_listener_class_discoverable(role, rolegroup_name)
+                })
                 .map(|(rolegroup_name, role_group)| {
                     (
                         self.rolegroup_ref(HbaseRole::Master.to_string(), rolegroup_name),
@@ -611,6 +620,9 @@ impl v1alpha1::HbaseCluster {
                 // Order rolegroups consistently, to avoid spurious downstream rewrites
                 .collect::<BTreeMap<_, _>>()
                 .into_iter()
+                .filter(|(rolegroup_name, _)| {
+                    self.resolved_listener_class_discoverable(role, rolegroup_name)
+                })
                 .map(|(rolegroup_name, role_group)| {
                     (
                         self.rolegroup_ref(HbaseRole::RegionServer.to_string(), rolegroup_name),
@@ -626,6 +638,9 @@ impl v1alpha1::HbaseCluster {
                 // Order rolegroups consistently, to avoid spurious downstream rewrites
                 .collect::<BTreeMap<_, _>>()
                 .into_iter()
+                .filter(|(rolegroup_name, _)| {
+                    self.resolved_listener_class_discoverable(role, rolegroup_name)
+                })
                 .map(|(rolegroup_name, role_group)| {
                     (
                         self.rolegroup_ref(HbaseRole::RestServer.to_string(), rolegroup_name),
@@ -636,6 +651,19 @@ impl v1alpha1::HbaseCluster {
         }
     }
 
+    fn resolved_listener_class_discoverable(
+        &self,
+        role: &HbaseRole,
+        rolegroup_name: &&String,
+    ) -> bool {
+        let listener_class = self.merged_listener_class(role, rolegroup_name);
+        if let Some(listener_class) = listener_class {
+            listener_class.discoverable()
+        } else {
+            false
+        }
+    }
+
     pub fn pod_refs(
         &self,
         role: &HbaseRole,
@@ -667,53 +695,113 @@ impl v1alpha1::HbaseCluster {
         &self,
         client: &stackable_operator::client::Client,
         role: &HbaseRole,
-        merged_config: &AnyServiceConfig,
         hbase_version: &str,
     ) -> Result<Vec<HbasePodRef>, Error> {
-        // only externally-reachable listeners are relevant
-        if merged_config.listener_class().discoverable() {
-            let pod_refs = self.pod_refs(role, hbase_version)?;
-            try_join_all(pod_refs.into_iter().map(|pod_ref| async {
-                // N.B. use the naming convention for persistent listener volumes as we
-                // have specified above that we only want externally-reachable endpoints.
-                let listener_name = format!("{LISTENER_VOLUME_NAME}-{}", pod_ref.pod_name);
-                let listener_ref =
-                    || ObjectRef::<Listener>::new(&listener_name).within(&pod_ref.namespace);
-                let pod_obj_ref =
-                    || ObjectRef::<Pod>::new(&pod_ref.pod_name).within(&pod_ref.namespace);
-                let listener = client
-                    .get::<Listener>(&listener_name, &pod_ref.namespace)
-                    .await
-                    .context(GetPodListenerSnafu {
-                        listener: listener_ref(),
-                        pod: pod_obj_ref(),
-                    })?;
-                let listener_address = listener
-                    .status
-                    .and_then(|s| s.ingress_addresses?.into_iter().next())
-                    .context(PodListenerHasNoAddressSnafu {
-                        listener: listener_ref(),
-                        pod: pod_obj_ref(),
-                    })?;
-                Ok(HbasePodRef {
-                    fqdn_override: Some(listener_address.address),
-                    ports: listener_address
-                        .ports
-                        .into_iter()
-                        .map(|(port_name, port)| {
-                            let port = u16::try_from(port).context(PortOutOfBoundsSnafu {
-                                port_name: &port_name,
-                                port,
-                            })?;
-                            Ok((port_name, port))
-                        })
-                        .collect::<Result<_, _>>()?,
-                    ..pod_ref
-                })
-            }))
-            .await
-        } else {
-            Ok(vec![])
+        let pod_refs = self.pod_refs(role, hbase_version)?;
+        try_join_all(pod_refs.into_iter().map(|pod_ref| async {
+            // N.B. use the naming convention for persistent listener volumes as we
+            // have specified above that we only want externally-reachable endpoints.
+            let listener_name = format!("{LISTENER_VOLUME_NAME}-{}", pod_ref.pod_name);
+            let listener_ref =
+                || ObjectRef::<Listener>::new(&listener_name).within(&pod_ref.namespace);
+            let pod_obj_ref =
+                || ObjectRef::<Pod>::new(&pod_ref.pod_name).within(&pod_ref.namespace);
+            let listener = client
+                .get::<Listener>(&listener_name, &pod_ref.namespace)
+                .await
+                .context(GetPodListenerSnafu {
+                    listener: listener_ref(),
+                    pod: pod_obj_ref(),
+                })?;
+            let listener_address = listener
+                .status
+                .and_then(|s| s.ingress_addresses?.into_iter().next())
+                .context(PodListenerHasNoAddressSnafu {
+                    listener: listener_ref(),
+                    pod: pod_obj_ref(),
+                })?;
+            Ok(HbasePodRef {
+                fqdn_override: Some(listener_address.address),
+                ports: listener_address
+                    .ports
+                    .into_iter()
+                    .map(|(port_name, port)| {
+                        let port = u16::try_from(port).context(PortOutOfBoundsSnafu {
+                            port_name: &port_name,
+                            port,
+                        })?;
+                        Ok((port_name, port))
+                    })
+                    .collect::<Result<_, _>>()?,
+                ..pod_ref
+            })
+        }))
+        .await
+    }
+
+    pub fn merged_listener_class(
+        &self,
+        role: &HbaseRole,
+        rolegroup_name: &String,
+    ) -> Option<SupportedListenerClasses> {
+        match role {
+            HbaseRole::Master => {
+                if let Some(masters) = self.spec.masters.as_ref() {
+                    let conf_defaults = Some(SupportedListenerClasses::ClusterInternal);
+                    let mut conf_role = masters.config.config.listener_class.to_owned();
+                    let mut conf_rolegroup = masters
+                        .role_groups
+                        .get(rolegroup_name)
+                        .map(|rg| rg.config.config.listener_class.clone())
+                        .unwrap_or_default();
+
+                    conf_role.merge(&conf_defaults);
+                    conf_rolegroup.merge(&conf_role);
+
+                    tracing::debug!("Merged listener-class: {:?} for {role}", conf_rolegroup);
+                    conf_rolegroup
+                } else {
+                    None
+                }
+            }
+            HbaseRole::RegionServer => {
+                if let Some(region_servers) = self.spec.region_servers.as_ref() {
+                    let conf_defaults = Some(SupportedListenerClasses::ClusterInternal);
+                    let mut conf_role = region_servers.config.config.listener_class.to_owned();
+                    let mut conf_rolegroup = region_servers
+                        .role_groups
+                        .get(rolegroup_name)
+                        .map(|rg| rg.config.config.listener_class.clone())
+                        .unwrap_or_default();
+
+                    conf_role.merge(&conf_defaults);
+                    conf_rolegroup.merge(&conf_role);
+
+                    tracing::debug!("Merged listener-class: {:?} for {role}", conf_rolegroup);
+                    conf_rolegroup
+                } else {
+                    None
+                }
+            }
+            HbaseRole::RestServer => {
+                if let Some(rest_servers) = self.spec.rest_servers.as_ref() {
+                    let conf_defaults = Some(SupportedListenerClasses::ClusterInternal);
+                    let mut conf_role = rest_servers.config.config.listener_class.to_owned();
+                    let mut conf_rolegroup = rest_servers
+                        .role_groups
+                        .get(rolegroup_name)
+                        .map(|rg| rg.config.config.listener_class.clone())
+                        .unwrap_or_default();
+
+                    conf_role.merge(&conf_defaults);
+                    conf_rolegroup.merge(&conf_role);
+
+                    tracing::debug!("Merged listener-class: {:?} for {role}", conf_rolegroup);
+                    conf_rolegroup
+                } else {
+                    None
+                }
+            }
         }
     }
 }

rust/operator-binary/src/discovery.rs

@@ -112,23 +112,19 @@ pub fn build_endpoint_configmap(
     );
 
     for role_podref in role_podrefs {
-        let role_name = role_podref.0;
-        // podrefs are written into the collection by replica index
-        // and can be retrieved in the same order
-        let mut i = 0;
         for podref in role_podref.1 {
             if let HbasePodRef {
                 fqdn_override: Some(fqdn_override),
                 ports,
+                pod_name,
                 ..
             } = podref
             {
                 if let Some(ui_port) = ports.get(&hbase.ui_port_name()) {
                     cmm.add_data(
-                        format!("hbase.{role_name}-{i}.ui"),
+                        format!("{pod_name}.http"),
                         format!("{fqdn_override}:{ui_port}"),
                     );
-                    i += 1;
                 }
             }
         }

rust/operator-binary/src/hbase_controller.rs

@@ -482,30 +482,6 @@ pub async fn reconcile_hbase(
                         rolegroup: rolegroup.clone(),
                     })?,
             );
-            // if the replicas are changed at the same time as the reconciliation
-            // being paused, it may be possible to have listeners that are *expected*
-            // (according to their replica number) but which are not yet created, so
-            // deactivate this action in such cases.
-            if hbase.spec.cluster_operation.reconciliation_paused
-                || hbase.spec.cluster_operation.stopped
-            {
-                tracing::info!(
-                    "Cluster is in a transitional state so do not attempt to collect listener information that will only be active once cluster has returned to a non-transitional state."
-                );
-            } else {
-                listener_refs.insert(
-                    hbase_role.to_string(),
-                    hbase
-                        .listener_refs(
-                            client,
-                            &hbase_role,
-                            &merged_config,
-                            &resolved_product_image.product_version,
-                        )
-                        .await
-                        .context(CollectDiscoveryConfigSnafu)?,
-                );
-            }
         }
 
         let role_config = hbase.role_config(&hbase_role);
@@ -517,6 +493,26 @@ pub async fn reconcile_hbase(
                 .await
                 .context(FailedToCreatePdbSnafu)?;
         }
+
+        // if the replicas are changed at the same time as the reconciliation
+        // being paused, it may be possible to have listeners that are *expected*
+        // (according to their replica number) but which are not yet created, so
+        // deactivate this action in such cases.
+        if hbase.spec.cluster_operation.reconciliation_paused
+            || hbase.spec.cluster_operation.stopped
+        {
+            tracing::info!(
+                "Cluster is in a transitional state so do not attempt to collect listener information that will only be active once cluster has returned to a non-transitional state."
+            );
+        } else {
+            listener_refs.insert(
+                hbase_role.to_string(),
+                hbase
+                    .listener_refs(client, &hbase_role, &resolved_product_image.product_version)
+                    .await
+                    .context(CollectDiscoveryConfigSnafu)?,
+            );
+        }
     }
 
     tracing::debug!(

tests/templates/kuttl/external-access/03-assert.yaml

@@ -16,13 +16,29 @@ kind: StatefulSet
 metadata:
   name: test-hbase-regionserver-default
 status:
-  readyReplicas: 2
-  replicas: 2
+  readyReplicas: 1
+  replicas: 1
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-hbase-regionserver-cluster-internal
+status:
+  readyReplicas: 1
+  replicas: 1
 ---
 apiVersion: apps/v1
 kind: StatefulSet
 metadata:
   name: test-hbase-restserver-default
 status:
-  readyReplicas: 2
-  replicas: 2
+  readyReplicas: 1
+  replicas: 1
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-hbase-restserver-external-unstable
+status:
+  readyReplicas: 1
+  replicas: 1

tests/templates/kuttl/external-access/03-install-hbase.yaml.j2

@@ -23,7 +23,7 @@ spec:
       gracefulShutdownTimeout: 1m
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
-      listenerClass: {{ test_scenario['values']['listener-class'] }}
+      listenerClass: external-unstable
     roleGroups:
       default:
         replicas: 2
@@ -32,16 +32,24 @@ spec:
       gracefulShutdownTimeout: 1m
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
-      listenerClass: {{ test_scenario['values']['listener-class'] }}
+      listenerClass: external-unstable
     roleGroups:
       default:
-        replicas: 2
+        replicas: 1
+      cluster-internal:
+        replicas: 1
+        config:
+          listenerClass: cluster-internal
   restServers:
     config:
       gracefulShutdownTimeout: 1m
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
-      listenerClass: {{ test_scenario['values']['listener-class'] }}
+      listenerClass: cluster-internal
     roleGroups:
       default:
-        replicas: 2
+        replicas: 1
+      external-unstable:
+        replicas: 1
+        config:
+          listenerClass: external-unstable