Reintroduce Endpoints-based fallback

Author: nightkr

Cargo.lock

@@ -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

Cargo.nix

@@ -3,7 +3,10 @@ use std::{
     sync::Arc,
 };
 
-use futures::{future::try_join_all, StreamExt};
+use futures::{
+    future::{try_join, try_join_all},
+    StreamExt,
+};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::OwnerReferenceBuilder,
@@ -12,7 +15,7 @@ use stackable_operator::{
         ListenerStatus, ServiceType,
     },
     k8s_openapi::{
-        api::core::v1::{Node, PersistentVolume, Service, ServicePort, ServiceSpec},
+        api::core::v1::{Endpoints, Node, PersistentVolume, Service, ServicePort, ServiceSpec},
         apimachinery::pkg::apis::meta::v1::LabelSelector,
     },
     kube::{
@@ -33,8 +36,28 @@ use stackable_operator::k8s_openapi::api::core::v1::Pod;
 const FIELD_MANAGER_SCOPE: &str = "listener";
 
 pub async fn run(client: stackable_operator::client::Client) {
-    controller::Controller::new(client.get_all_api::<Listener>(), watcher::Config::default())
+    let controller =
+        controller::Controller::new(client.get_all_api::<Listener>(), watcher::Config::default());
+    let listener_store = controller.store();
+    controller
         .owns(client.get_all_api::<Service>(), watcher::Config::default())
+        .watches(
+            client.get_all_api::<Endpoints>(),
+            watcher::Config::default(),
+            move |endpoints| {
+                listener_store
+                    .state()
+                    .into_iter()
+                    .filter(move |listener| {
+                        listener
+                            .status
+                            .as_ref()
+                            .and_then(|s| s.service_name.as_deref())
+                            == endpoints.metadata.name.as_deref()
+                    })
+                    .map(|l| ObjectRef::from_obj(&*l))
+            },
+        )
         .watches(
             client.get_all_api::<PersistentVolume>(),
             watcher::Config::default(),
@@ -81,6 +104,10 @@ pub enum Error {
     ListenerPodSelector {
         source: ListenerMountedPodLabelError,
     },
+    #[snafu(display("failed to get PersistentVolumes for Listener"))]
+    GetListenerPvs {
+        source: stackable_operator::client::Error,
+    },
     #[snafu(display("failed to get {obj}"))]
     GetObject {
         source: stackable_operator::client::Error,
@@ -113,6 +140,7 @@ impl ReconcilerError for Error {
             Self::NoListenerClass => None,
             Self::ListenerPvSelector { source: _ } => None,
             Self::ListenerPodSelector { source: _ } => None,
+            Self::GetListenerPvs { source: _ } => None,
             Self::GetObject { source: _, obj } => Some(obj.clone()),
             Self::BuildListenerOwnerRef { .. } => None,
             Self::ApplyService { source: _, svc } => Some(svc.clone().erase()),
@@ -220,18 +248,34 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
     let ports: BTreeMap<String, i32>;
     match listener_class.spec.service_type {
         ServiceType::NodePort => {
-            let pvs = ctx
-                .client
-                .list_with_label_selector::<PersistentVolume>(
-                    &(),
-                    &LabelSelector {
-                        match_labels: Some(listener_persistent_volume_label(&listener).unwrap()),
-                        ..Default::default()
-                    },
-                )
-                .await
-                .unwrap();
-            let node_names = pvs
+            let (pvs, endpoints) = try_join(
+                async {
+                    ctx.client
+                        .list_with_label_selector::<PersistentVolume>(
+                            &(),
+                            &LabelSelector {
+                                match_labels: Some(
+                                    listener_persistent_volume_label(&listener).unwrap(),
+                                ),
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .context(GetListenerPvsSnafu)
+                },
+                async {
+                    ctx.client
+                        // Endpoints object may not yet be created by its respective controller
+                        .get_opt::<Endpoints>(&svc_name, ns)
+                        .await
+                        .with_context(|_| GetObjectSnafu {
+                            obj: ObjectRef::<Endpoints>::new(&svc_name).within(ns).erase(),
+                        })
+                },
+            )
+            .await?;
+
+            let pv_node_names = pvs
                 .into_iter()
                 .filter_map(|pv| pv.spec?.node_affinity?.required)
                 .flat_map(|affinity| affinity.node_selector_terms)
@@ -241,6 +285,33 @@ pub async fn reconcile(listener: Arc<Listener>, ctx: Arc<Ctx>) -> Result<control
                 .filter_map(|expr| expr.values)
                 .flatten()
                 .collect::<BTreeSet<_>>();
+
+            // Old objects that haven't been mounted before the PV lookup mechanism was added will
+            // not have the correct labels, so we also look up using Endpoints.
+            let endpoints_node_names = endpoints
+                .into_iter()
+                .filter_map(|endpoints| endpoints.subsets)
+                .flatten()
+                .flat_map(|subset| subset.addresses)
+                .flatten()
+                .flat_map(|addr| addr.node_name)
+                .collect::<BTreeSet<_>>();
+
+            let node_names_missing_from_pv = endpoints_node_names
+                .difference(&pv_node_names)
+                .collect::<Vec<_>>();
+            if !node_names_missing_from_pv.is_empty() {
+                tracing::warn!(
+                    ?node_names_missing_from_pv,
+                    "some backing Nodes could only be found via legacy Endpoints discovery method, {} {}",
+                    "this may cause discovery config to be unstable",
+                    "(hint: try restarting the Pods backing this Listener)"
+                );
+            }
+
+            let mut node_names = pv_node_names;
+            node_names.extend(endpoints_node_names);
+
             nodes = try_join_all(node_names.iter().map(|node_name| async {
                 ctx.client
                     .get::<Node>(node_name, &())