[CI-1951] fix(podiprecovery): add Pod watch so recovery is level-triggered

pkg/controller/podiprecovery/podiprecovery_controller.go

@@ -33,6 +33,15 @@
 // spec.hostNetwork == true as a side effect of the normal apply path.
 // The controller additionally verifies spec.hostNetwork == true on each
 // candidate as a safety net before deleting.
+//
+// Recovery is triggered by two watches so it stays level-triggered rather
+// than reacting to a single node event: a Node watch (fires when a node's
+// host IPs change) and a Pod watch (fires when a host-networked pod finishes
+// (re)starting). The Pod watch is what catches a pod that was still
+// restarting at the moment its node's IP changed and only later comes back
+// reporting its old, stale IP — after which no further Node event would
+// fire. Both watches enqueue the same Node key and share one idempotent
+// Reconcile.
 package podiprecovery
 
 import (
@@ -41,6 +50,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -83,9 +93,34 @@ func Add(mgr manager.Manager, _ options.ControllerOptions) error {
 		return fmt.Errorf("podiprecovery-controller failed to watch Nodes: %w", err)
 	}
 
+	// Watch host-networked pods too, and re-enqueue the pod's node when one
+	// settles into a Running-with-IPs state. The Node watch catches the
+	// instant a node's IP changes, but a pod that is still restarting at
+	// that instant has no status.podIPs yet and is skipped by Reconcile;
+	// when the kubelet finishes (re)starting it, the surviving pod reports
+	// its old (now stale) IP and no further Node event ever fires. Without
+	// this second watch that late-settling pod would never be re-evaluated
+	// (the recovery would be edge-triggered on the node event alone). The
+	// map function keys back to the Node, and Reconcile is idempotent, so
+	// the extra enqueues for already-healthy pods are cheap no-ops.
+	if err := c.WatchObject(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(podToNode), hostNetPodSettledPredicate()); err != nil {
+		return fmt.Errorf("podiprecovery-controller failed to watch Pods: %w", err)
+	}
+
 	return nil
 }
 
+// podToNode maps a Pod event to a reconcile request for the Node the pod
+// runs on. Recovery is keyed on the Node, so a settling pod triggers a full
+// re-evaluation of every host-networked pod on its node.
+func podToNode(_ context.Context, obj client.Object) []reconcile.Request {
+	pod, ok := obj.(*corev1.Pod)
+	if !ok || pod.Spec.NodeName == "" {
+		return nil
+	}
+	return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: pod.Spec.NodeName}}}
+}
+
 // hostIPsChangedPredicate filters Node events so reconciles only fire when
 // the node's host IPs change (including initial set / removal). New nodes
 // are reconciled once to handle the case where pods are scheduled before the
@@ -112,6 +147,73 @@ func hostIPsChangedPredicate() predicate.Predicate {
 	}
 }
 
+// hostNetPodSettledPredicate filters Pod events down to operator-managed
+// hostNetwork pods that have just settled into a state where their
+// status.podIPs can be meaningfully compared against their node's host IPs.
+//
+// This is the second half of the recovery trigger (see the Pod watch in
+// Add). It fires when such a pod finishes (re)starting — either its reported
+// IPs change, or it transitions to Ready — which is the moment a pod that
+// survived a node-IP change comes back reporting its old, now-stale IP.
+func hostNetPodSettledPredicate() predicate.Predicate {
+	return predicate.Funcs{
+		CreateFunc: func(e event.CreateEvent) bool {
+			pod, ok := e.Object.(*corev1.Pod)
+			// An already-running host-net pod observed on (re)start of the
+			// controller: evaluate it if it already reports IPs.
+			return ok && isManagedHostNetPod(pod) && len(pod.Status.PodIPs) > 0
+		},
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			oldPod, oldOK := e.ObjectOld.(*corev1.Pod)
+			newPod, newOK := e.ObjectNew.(*corev1.Pod)
+			if !oldOK || !newOK {
+				return false
+			}
+			if !isManagedHostNetPod(newPod) || len(newPod.Status.PodIPs) == 0 {
+				return false
+			}
+			// Re-evaluate when the pod's reported IPs change or when it
+			// transitions to Ready — both mark the point at which a
+			// (re)started pod's final status.podIPs becomes observable.
+			if !podIPSet(oldPod).Equal(podIPSet(newPod)) {
+				return true
+			}
+			return !podReady(oldPod) && podReady(newPod)
+		},
+		DeleteFunc: func(e event.DeleteEvent) bool {
+			return false
+		},
+		GenericFunc: func(e event.GenericEvent) bool {
+			return false
+		},
+	}
+}
+
+// isManagedHostNetPod reports whether the pod is an operator-managed
+// hostNetwork pod — the only kind the recovery controller acts on.
+func isManagedHostNetPod(pod *corev1.Pod) bool {
+	return pod.Spec.HostNetwork && pod.Labels[common.HostNetworkedPodLabel] == "true"
+}
+
+// podIPSet returns the set of IPs reported in the pod's status.podIPs.
+func podIPSet(pod *corev1.Pod) sets.Set[string] {
+	out := sets.New[string]()
+	for _, pip := range pod.Status.PodIPs {
+		out.Insert(pip.IP)
+	}
+	return out
+}
+
+// podReady reports whether the pod's Ready condition is currently True.
+func podReady(pod *corev1.Pod) bool {
+	for _, c := range pod.Status.Conditions {
+		if c.Type == corev1.PodReady {
+			return c.Status == corev1.ConditionTrue
+		}
+	}
+	return false
+}
+
 // nodeHostIPSet returns the set of addresses that the kubelet would use to
 // populate `status.podIPs` for a hostNetwork pod scheduled on this node.
 //

pkg/controller/podiprecovery/podiprecovery_controller_test.go

@@ -335,6 +335,102 @@ var _ = Describe("PodIPRecovery controller", func() {
 			Expect(pred.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: new})).To(BeTrue())
 		})
 	})
+
+	Context("hostNetPodSettledPredicate", func() {
+		pred := hostNetPodSettledPredicate()
+
+		// settledPod builds a managed host-networked pod on node1 with the
+		// given IPs and readiness. Tests override individual fields as needed.
+		settledPod := func(podIPs []string, ready bool) *corev1.Pod {
+			p := &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "p1",
+					Namespace: ns,
+					Labels:    map[string]string{common.HostNetworkedPodLabel: "true"},
+				},
+				Spec: corev1.PodSpec{NodeName: "node1", HostNetwork: true},
+			}
+			for _, ip := range podIPs {
+				p.Status.PodIPs = append(p.Status.PodIPs, corev1.PodIP{IP: ip})
+			}
+			cond := corev1.ConditionFalse
+			if ready {
+				cond = corev1.ConditionTrue
+			}
+			p.Status.Conditions = []corev1.PodCondition{{Type: corev1.PodReady, Status: cond}}
+			return p
+		}
+
+		It("enqueues on Create when a managed host-net pod already reports IPs", func() {
+			Expect(pred.Create(event.CreateEvent{Object: settledPod([]string{"10.0.0.1"}, true)})).To(BeTrue())
+		})
+
+		It("does not enqueue on Create when the pod has no IPs yet", func() {
+			Expect(pred.Create(event.CreateEvent{Object: settledPod(nil, false)})).To(BeFalse())
+		})
+
+		It("does not enqueue on Create for a pod without the host-net marker label", func() {
+			p := settledPod([]string{"10.0.0.1"}, true)
+			p.Labels = nil
+			Expect(pred.Create(event.CreateEvent{Object: p})).To(BeFalse())
+		})
+
+		It("does not enqueue on Create for a non-hostNetwork pod", func() {
+			p := settledPod([]string{"10.0.0.1"}, true)
+			p.Spec.HostNetwork = false
+			Expect(pred.Create(event.CreateEvent{Object: p})).To(BeFalse())
+		})
+
+		It("enqueues on Update when the pod's IPs appear (empty -> set)", func() {
+			old := settledPod(nil, false)
+			new := settledPod([]string{"10.0.0.1"}, false)
+			Expect(pred.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: new})).To(BeTrue())
+		})
+
+		It("enqueues on Update when the pod becomes Ready (this is the late-settling survivor case)", func() {
+			// The surviving pod comes back reporting its old, now-stale IP;
+			// its IPs don't change but it transitions to Ready. This is the
+			// exact edge the Node watch alone would miss.
+			old := settledPod([]string{"10.0.0.1"}, false)
+			new := settledPod([]string{"10.0.0.1"}, true)
+			Expect(pred.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: new})).To(BeTrue())
+		})
+
+		It("does not enqueue on Update when nothing relevant changed (steady-state heartbeat)", func() {
+			old := settledPod([]string{"10.0.0.1"}, true)
+			new := settledPod([]string{"10.0.0.1"}, true)
+			Expect(pred.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: new})).To(BeFalse())
+		})
+
+		It("does not enqueue on Update for a pod without the host-net marker label", func() {
+			old := settledPod(nil, false)
+			old.Labels = nil
+			new := settledPod([]string{"10.0.0.1"}, true)
+			new.Labels = nil
+			Expect(pred.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: new})).To(BeFalse())
+		})
+
+		It("does not enqueue on Delete", func() {
+			Expect(pred.Delete(event.DeleteEvent{Object: settledPod([]string{"10.0.0.1"}, true)})).To(BeFalse())
+		})
+	})
+
+	Context("podToNode", func() {
+		It("maps a pod to a reconcile request for its node", func() {
+			pod := newPod("p1", "node7", "10.0.0.1", true)
+			reqs := podToNode(ctx, pod)
+			Expect(reqs).To(ConsistOf(reconcile.Request{NamespacedName: types.NamespacedName{Name: "node7"}}))
+		})
+
+		It("returns nothing for a pod not yet scheduled to a node", func() {
+			pod := newPod("p1", "", "10.0.0.1", true)
+			Expect(podToNode(ctx, pod)).To(BeEmpty())
+		})
+
+		It("returns nothing for a non-pod object", func() {
+			Expect(podToNode(ctx, newNode("node1", "10.0.0.1"))).To(BeEmpty())
+		})
+	})
 })
 
 // newPodWithLabels is the lower-level helper used by the `newPod` shortcut.