diff --git a/pkg/controller/podiprecovery/podiprecovery_controller.go b/pkg/controller/podiprecovery/podiprecovery_controller.go index b2a2e0899c..2f8c8469da 100644 --- a/pkg/controller/podiprecovery/podiprecovery_controller.go +++ b/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. // diff --git a/pkg/controller/podiprecovery/podiprecovery_controller_test.go b/pkg/controller/podiprecovery/podiprecovery_controller_test.go index d1ea9de9bf..7e262cc344 100644 --- a/pkg/controller/podiprecovery/podiprecovery_controller_test.go +++ b/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.