Use address type (IP vs DNS) configured by ListenerClass

CHANGELOG.md

@@ -4,16 +4,22 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Listener.status.addresses now allows users to configure whether they prefer IP addresses or DNS hostnames ([#233]).
+
 ### Changed
 
 - Listener.status.addresses for NodePort listeners now includes replicas that are currently unavailable ([#231]).
+- Listener.status.addresses now defaults to DNS hostnames for all service types (previously NodePort and ClusterIP would prefer IP addresses, [#233]).
 
 ### Fixed
 
 - Listener.status.addresses is now de-duplicated ([#231]).
 - Listener controller now listens for ListenerClass updates ([#231]).
 
 [#231]: https://github.com/stackabletech/listener-operator/pull/231
+[#233]: https://github.com/stackabletech/listener-operator/pull/233
 
 ## [24.7.0] - 2024-07-24
 

Cargo.toml

@@ -21,7 +21,7 @@ prost = "0.13"
 prost-types = "0.13"
 serde = "1.0"
 snafu = "0.8"
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.76.0" }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.78.0" }
 strum = { version = "0.26", features = ["derive"] }
 socket2 = { version = "0.5", features = ["all"] }
 tokio = { version = "1.40", features = ["full"] }
@@ -31,5 +31,7 @@ tonic-build = "0.12"
 tonic-reflection = "0.12"
 tracing = "0.1.40"
 
-# [patch."https://github.com/stackabletech/operator-rs.git"]
+[patch."https://github.com/stackabletech/operator-rs.git"]
+# stackable-operator = { path = "../operator-rs/crates/stackable-operator" }
 # stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main" }
+stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "feature/listenerclass-preferred-address-type" }

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

@@ -24,6 +24,16 @@ spec:
             spec:
               description: Defines a policy for how [Listeners](https://docs.stackable.tech/home/nightly/listener-operator/listener) should be exposed. Read the [ListenerClass documentation](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass) for more information.
               properties:
+                preferredAddressType:
+                  default: Hostname
+                  description: |-
+                    Whether addresses should prefer using the IP address (`IP`) or the hostname (`Hostname`).
+
+                    The other type will be used if the preferred type is not available.
+                  enum:
+                    - Hostname
+                    - IP
+                  type: string
                 serviceAnnotations:
                   additionalProperties:
                     type: string

rust/operator-binary/src/csi_server/node.rs

@@ -4,8 +4,8 @@ use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::OwnerReferenceBuilder,
     commons::listener::{
-        Listener, ListenerIngress, ListenerPort, ListenerSpec, PodListener, PodListenerScope,
-        PodListeners, PodListenersSpec,
+        Listener, ListenerClass, ListenerIngress, ListenerPort, ListenerSpec, PodListener,
+        PodListenerScope, PodListeners, PodListenersSpec,
     },
     k8s_openapi::api::core::v1::{Node, PersistentVolume, PersistentVolumeClaim, Pod, Volume},
     kube::{
@@ -72,6 +72,9 @@ enum PublishVolumeError {
         listener: ObjectRef<Listener>,
     },
 
+    #[snafu(display("{listener} has no associated ListenerClass"))]
+    ListenerHasNoClass { listener: ObjectRef<Listener> },
+
     #[snafu(display("{pod} has not been scheduled to a node yet"))]
     PodHasNoNode { pod: ObjectRef<Pod> },
 
@@ -131,6 +134,7 @@ impl From<PublishVolumeError> for Status {
             PublishVolumeError::PodHasNoNode { .. } => Status::unavailable(full_msg),
             PublishVolumeError::ListenerPvReference { .. } => Status::failed_precondition(full_msg),
             PublishVolumeError::ListenerPodSelector { .. } => Status::failed_precondition(full_msg),
+            PublishVolumeError::ListenerHasNoClass { .. } => Status::failed_precondition(full_msg),
             PublishVolumeError::BuildListenerOwnerRef { .. } => Status::unavailable(full_msg),
             PublishVolumeError::ApplyListener { .. } => Status::unavailable(full_msg),
             PublishVolumeError::AddListenerLabelToPv { .. } => Status::unavailable(full_msg),
@@ -442,10 +446,25 @@ async fn local_listener_addresses_for_pod(
             .get::<Node>(node_name, &())
             .await
             .with_context(|_| GetObjectSnafu {
-                obj: { ObjectRef::<Node>::new(node_name).erase() },
+                obj: ObjectRef::<Node>::new(node_name).erase(),
+            })?;
+        let listener_class_name =
+            listener
+                .spec
+                .class_name
+                .as_deref()
+                .with_context(|| ListenerHasNoClassSnafu {
+                    listener: ObjectRef::from_obj(listener),
+                })?;
+        let listener_class = client
+            .get::<ListenerClass>(listener_class_name, &())
+            .await
+            .with_context(|_| GetObjectSnafu {
+                obj: ObjectRef::<ListenerClass>::new(listener_class_name).erase(),
             })?;
 
         Ok(node_primary_address(&node)
+            .pick(listener_class.spec.preferred_address_type)
             .map(|(address, address_type)| ListenerIngress {
                 // nodes: Some(vec![node_name.to_string()]),
                 address: address.to_string(),

rust/operator-binary/src/listener_controller.rs

@@ -28,7 +28,10 @@ use stackable_operator::{
 };
 use strum::IntoStaticStr;
 
-use crate::{csi_server::node::NODE_TOPOLOGY_LABEL_HOSTNAME, utils::node_primary_address};
+use crate::{
+    csi_server::node::NODE_TOPOLOGY_LABEL_HOSTNAME,
+    utils::{node_primary_address, AddressCandidates},
+};
 
 #[cfg(doc)]
 use stackable_operator::k8s_openapi::api::core::v1::Pod;
@@ -269,6 +272,7 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
         })?;
 
     let nodes: Vec<Node>;
+    let kubernetes_service_fqdn: String;
     let addresses: Vec<(&str, AddressType)>;
     let ports: BTreeMap<String, i32>;
     match listener_class.spec.service_type {
@@ -286,7 +290,9 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
             .await?;
             addresses = nodes
                 .iter()
-                .flat_map(node_primary_address)
+                .flat_map(|node| {
+                    node_primary_address(node).pick(listener_class.spec.preferred_address_type)
+                })
                 .collect::<Vec<_>>();
             ports = svc
                 .spec
@@ -304,11 +310,11 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
                 .flat_map(|ss| ss.load_balancer.as_ref()?.ingress.as_ref())
                 .flatten()
                 .flat_map(|ingress| {
-                    ingress
-                        .hostname
-                        .as_deref()
-                        .zip(Some(AddressType::Hostname))
-                        .or_else(|| ingress.ip.as_deref().zip(Some(AddressType::Ip)))
+                    AddressCandidates {
+                        ip: ingress.ip.as_deref(),
+                        hostname: ingress.hostname.as_deref(),
+                    }
+                    .pick(listener_class.spec.preferred_address_type)
                 })
                 .collect();
             ports = svc
@@ -321,13 +327,19 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
                 .collect();
         }
         ServiceType::ClusterIP => {
-            addresses = svc
-                .spec
-                .iter()
-                .flat_map(|s| &s.cluster_ips)
-                .flatten()
-                .map(|addr| (&**addr, AddressType::Ip))
-                .collect::<Vec<_>>();
+            addresses = match listener_class.spec.preferred_address_type {
+                AddressType::Ip => svc
+                    .spec
+                    .iter()
+                    .flat_map(|s| &s.cluster_ips)
+                    .flatten()
+                    .map(|addr| (&**addr, AddressType::Ip))
+                    .collect::<Vec<_>>(),
+                AddressType::Hostname => {
+                    kubernetes_service_fqdn = format!("{svc_name}.{ns}.svc.cluster.local");
+                    vec![(&kubernetes_service_fqdn, AddressType::Hostname)]
+                }
+            };
             ports = svc
                 .spec
                 .as_ref()