Generate NodePort address listings from PersistentVolume reservations

CHANGELOG.md

@@ -4,6 +4,17 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Changed
+
+- Listener.status.addresses for NodePort listeners now includes replicas that are currently unavailable ([#231]).
+
+### 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
+
 ## [24.7.0] - 2024-07-24
 
 ### Added

rust/csi-grpc/src/lib.rs

@@ -1,5 +1,8 @@
 //! Include gRPC definition files that have been generated by `build.rs`
 
+// CSI docs don't quite align with Rustdoc conventions
+#![allow(clippy::doc_lazy_continuation)]
+
 pub static FILE_DESCRIPTOR_SET_BYTES: &[u8] =
     include_bytes!(concat!(env!("OUT_DIR"), "/file_descriptor_set.bin"));
 

rust/operator-binary/Cargo.toml

@@ -27,6 +27,7 @@ serde.workspace = true
 snafu.workspace = true
 strum.workspace = true
 h2.workspace = true
+tracing = "0.1.40"
 
 [build-dependencies]
 built.workspace = true

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

@@ -17,14 +17,19 @@ use std::{fmt::Debug, path::PathBuf};
 use tonic::{Request, Response, Status};
 
 use crate::{
-    listener_controller::{listener_mounted_pod_label, ListenerMountedPodLabelError},
+    listener_controller::{
+        listener_mounted_pod_label, listener_persistent_volume_label, ListenerMountedPodLabelError,
+        ListenerPersistentVolumeLabelError,
+    },
     utils::{error_full_message, node_primary_address},
 };
 
 use super::{tonic_unimplemented, ListenerSelector, ListenerVolumeContext};
 
 const FIELD_MANAGER_SCOPE: &str = "volume";
 
+pub const NODE_TOPOLOGY_LABEL_HOSTNAME: &str = "listeners.stackable.tech/hostname";
+
 pub struct ListenerOperatorNode {
     pub client: stackable_operator::client::Client,
     pub node_name: String,
@@ -55,6 +60,12 @@ enum PublishVolumeError {
     #[snafu(display("PersistentVolume has no corresponding PersistentVolumeClaim"))]
     UnclaimedPv,
 
+    #[snafu(display("failed to generate {listener}'s PersistentVolume selector"))]
+    ListenerPvReference {
+        source: ListenerPersistentVolumeLabelError,
+        listener: ObjectRef<Listener>,
+    },
+
     #[snafu(display("failed to generate {listener}'s pod selector"))]
     ListenerPodSelector {
         source: ListenerMountedPodLabelError,
@@ -75,6 +86,12 @@ enum PublishVolumeError {
         listener: ObjectRef<Listener>,
     },
 
+    #[snafu(display("failed to add listener label to {pv}"))]
+    AddListenerLabelToPv {
+        source: stackable_operator::client::Error,
+        pv: ObjectRef<PersistentVolume>,
+    },
+
     #[snafu(display("failed to add listener label to {pod}"))]
     AddListenerLabelToPod {
         source: stackable_operator::client::Error,
@@ -112,9 +129,11 @@ impl From<PublishVolumeError> for Status {
             PublishVolumeError::GetObject { .. } => Status::unavailable(full_msg),
             PublishVolumeError::UnclaimedPv => Status::unavailable(full_msg),
             PublishVolumeError::PodHasNoNode { .. } => Status::unavailable(full_msg),
+            PublishVolumeError::ListenerPvReference { .. } => Status::failed_precondition(full_msg),
             PublishVolumeError::ListenerPodSelector { .. } => Status::failed_precondition(full_msg),
             PublishVolumeError::BuildListenerOwnerRef { .. } => Status::unavailable(full_msg),
             PublishVolumeError::ApplyListener { .. } => Status::unavailable(full_msg),
+            PublishVolumeError::AddListenerLabelToPv { .. } => Status::unavailable(full_msg),
             PublishVolumeError::AddListenerLabelToPod { .. } => Status::unavailable(full_msg),
             PublishVolumeError::NoAddresses { .. } => Status::unavailable(full_msg),
             PublishVolumeError::PreparePodDir { .. } => Status::internal(full_msg),
@@ -155,7 +174,7 @@ impl csi::v1::node_server::Node for ListenerOperatorNode {
             max_volumes_per_node: i64::MAX,
             accessible_topology: Some(Topology {
                 segments: [(
-                    "listeners.stackable.tech/hostname".to_string(),
+                    NODE_TOPOLOGY_LABEL_HOSTNAME.to_string(),
                     self.node_name.clone(),
                 )]
                 .into(),
@@ -276,6 +295,29 @@ impl csi::v1::node_server::Node for ListenerOperatorNode {
             }
         };
 
+        // Add listener label to PV, allowing traffic to be directed based on reservations, rather than which replicas are *currently* active.
+        // See https://github.com/stackabletech/listener-operator/issues/220
+        self.client
+            .apply_patch(
+                FIELD_MANAGER_SCOPE,
+                &pv,
+                &PersistentVolume {
+                    metadata: ObjectMeta {
+                        labels: Some(listener_persistent_volume_label(&listener).context(
+                            ListenerPvReferenceSnafu {
+                                listener: ObjectRef::from_obj(&listener),
+                            },
+                        )?),
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                },
+            )
+            .await
+            .with_context(|_| AddListenerLabelToPvSnafu {
+                pv: ObjectRef::from_obj(&pv),
+            })?;
+
         // Add listener label to pod so that traffic can be directed to it
         self.client
             // IMPORTANT