fix: prevent premature EgressIP assignment to br-ex interface

EgressIPs were incorrectly added to the br-ex network interface immediately
upon EgressIP CR creation, without checking for matching pods or using
longest prefix matching (LPM) for interface selection. This caused duplicate
IP assignments when the correct secondary host interface was later assigned.

Changes:
- Add pod existence validation before IP assignment in bridgeEIPAddrManager
- Implement longest prefix matching to determine correct network interface
- Add dynamic pod event handling to re-evaluate EgressIPs on pod lifecycle
- Extend retry framework to watch pod events for EgressIP reconciliation
- Add comprehensive unit tests covering edge cases and pod matching logic

The fix ensures EgressIPs are only assigned when:
1. Matching pods exist and are running
2. The target interface is determined via LPM algorithm
3. No duplicate IPs are assigned across multiple interfaces

Files modified:
- gateway_egressip.go: Core logic for pod checking and LPM interface selection
- gateway.go: Pod event handling and retry framework integration
- gateway_shared_intf.go: Bridge manager initialization with pod informers
- obj_retry_gateway.go: Pod resource handling in retry framework
- gateway_egressip_test.go: Comprehensive test coverage for new functionality

Fixes duplicate EgressIP assignment bug where IPs appeared on both br-ex
and the correct secondary interface simultaneously.

Author: cgoncalves

go-controller/pkg/node/egressip/gateway_egressip.go

@@ -5,18 +5,22 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"net/netip"
 	"sync"
 
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	corev1informers "k8s.io/client-go/informers/core/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 
+	"github.com/gaissmai/cidrtree"
 	egressipv1 "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressip/v1"
 	egressipinformers "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressip/v1/apis/informers/externalversions/egressip/v1"
 	egressiplisters "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/crd/egressip/v1/apis/listers/egressip/v1"
@@ -173,13 +177,15 @@ func (mic *MarkIPsCache) setSyncdOnce() {
 	mic.mu.Unlock()
 }
 
-type BridgeEIPAddrManager struct {
+type bridgeEIPAddrManager struct {
 	nodeName         string
 	bridgeName       string
 	nodeAnnotationMu sync.Mutex
 	eIPLister        egressiplisters.EgressIPLister
 	eIPInformer      cache.SharedIndexInformer
 	nodeLister       corev1listers.NodeLister
+	podLister        corev1listers.PodLister
+	namespaceLister  corev1listers.NamespaceLister
 	kube             kube.Interface
 	addrManager      *linkmanager.Controller
 	cache            *MarkIPsCache
@@ -190,31 +196,173 @@ type BridgeEIPAddrManager struct {
 // prior to restarting.
 // It provides the assigned IPs info node IP handler. Node IP handler must not consider assigned EgressIP IPs as possible node IPs.
 // Openflow manager must generate the SNAT openflow conditional on packet marks and therefore needs access to EIP IPs and associated packet marks.
-// BridgeEIPAddrManager must be able to force Openflow manager to resync if EgressIP assignment for the node changes.
-func NewBridgeEIPAddrManager(nodeName, bridgeName string, linkManager *linkmanager.Controller,
-	kube kube.Interface, eIPInformer egressipinformers.EgressIPInformer, nodeInformer corev1informers.NodeInformer) *BridgeEIPAddrManager {
-	return &BridgeEIPAddrManager{
+// bridgeEIPAddrManager must be able to force Openflow manager to resync if EgressIP assignment for the node changes.
+func newBridgeEIPAddrManager(nodeName, bridgeName string, linkManager *linkmanager.Controller,
+	kube kube.Interface, eIPInformer egressipinformers.EgressIPInformer, nodeInformer corev1informers.NodeInformer,
+	podInformer corev1informers.PodInformer, namespaceInformer corev1informers.NamespaceInformer) *bridgeEIPAddrManager {
+	return &bridgeEIPAddrManager{
 		nodeName:         nodeName,     // k8 node name
 		bridgeName:       bridgeName,   // bridge name for which EIP IPs are managed
 		nodeAnnotationMu: sync.Mutex{}, // mu for updating Node annotation
 		eIPLister:        eIPInformer.Lister(),
 		eIPInformer:      eIPInformer.Informer(),
 		nodeLister:       nodeInformer.Lister(),
+		podLister:        podInformer.Lister(),
+		namespaceLister:  namespaceInformer.Lister(),
 		kube:             kube,
 		addrManager:      linkManager,
 		cache:            NewMarkIPsCache(), // cache to store pkt mark -> EIP IP.
 	}
 }
 
-func (g *BridgeEIPAddrManager) GetCache() *MarkIPsCache {
+func (g *bridgeEIPAddrManager) GetCache() *MarkIPsCache {
 	return g.cache
 }
 
-func (g *BridgeEIPAddrManager) AddEgressIP(eip *egressipv1.EgressIP) (bool, error) {
+// NewBridgeEIPAddrManager creates a new bridge EIP address manager
+func NewBridgeEIPAddrManager(nodeName, bridgeName string, linkManager *linkmanager.Controller,
+	kube kube.Interface, eIPInformer egressipinformers.EgressIPInformer, nodeInformer corev1informers.NodeInformer) *bridgeEIPAddrManager {
+	return newBridgeEIPAddrManager(nodeName, bridgeName, linkManager, kube, eIPInformer, nodeInformer, nil, nil)
+}
+
+// findLinkOnSameNetworkAsIPUsingLPM finds the correct interface for an EgressIP using longest prefix match
+// This is based on the implementation in the secondary EgressIP controller
+func (g *bridgeEIPAddrManager) findLinkOnSameNetworkAsIPUsingLPM(ip net.IP, v4, v6 bool) (bool, netlink.Link, error) {
+	prefixLinks := map[string]netlink.Link{} // key is network CIDR
+	prefixes := make([]netip.Prefix, 0)
+	links, err := util.GetNetLinkOps().LinkList()
+	if err != nil {
+		return false, nil, fmt.Errorf("failed to list links: %v", err)
+	}
+	for _, link := range links {
+		link := link
+		linkPrefixes, err := g.getFilteredPrefixes(link, v4, v6)
+		if err != nil {
+			klog.Errorf("Failed to get address from link %s: %v", link.Attrs().Name, err)
+			continue
+		}
+		prefixes = append(prefixes, linkPrefixes...)
+		// create lookup table for later retrieval
+		for _, prefixFound := range linkPrefixes {
+			_, ipNet, err := net.ParseCIDR(prefixFound.String())
+			if err != nil {
+				klog.Errorf("Egress IP: skipping prefix %q due to parsing CIDR error: %v", prefixFound.String(), err)
+				continue
+			}
+			prefixLinks[ipNet.String()] = link
+		}
+	}
+	lpmTree := cidrtree.New(prefixes...)
+	addr, err := netip.ParseAddr(ip.String())
+	if err != nil {
+		return false, nil, fmt.Errorf("failed to convert IP %s to netip addr: %v", ip.String(), err)
+	}
+	network, found := lpmTree.Lookup(addr)
+	if !found {
+		return false, nil, nil
+	}
+	link, ok := prefixLinks[network.String()]
+	if !ok {
+		return false, nil, nil
+	}
+	return true, link, nil
+}
+
+// getFilteredPrefixes returns address Prefixes from interfaces with filtering
+func (g *bridgeEIPAddrManager) getFilteredPrefixes(link netlink.Link, v4, v6 bool) ([]netip.Prefix, error) {
+	validAddresses := make([]netip.Prefix, 0)
+	flags := link.Attrs().Flags.String()
+	if !g.isLinkUp(flags) {
+		return validAddresses, nil
+	}
+	linkAddresses, err := util.GetFilteredInterfaceAddrs(link, v4, v6)
+	if err != nil {
+		return validAddresses, err
+	}
+	for _, addr := range linkAddresses {
+		// Skip single-host addresses (/32 for IPv4, /128 for IPv6)
+		ones, bits := addr.Mask.Size()
+		if ones == bits {
+			continue
+		}
+		// Convert to netip.Prefix
+		prefix, err := netip.ParsePrefix(addr.String())
+		if err != nil {
+			klog.Errorf("Failed to parse address %s as netip.Prefix: %v", addr.String(), err)
+			continue
+		}
+		validAddresses = append(validAddresses, prefix)
+	}
+	return validAddresses, nil
+}
+
+// isLinkUp checks if a network link is up
+func (g *bridgeEIPAddrManager) isLinkUp(flags string) bool {
+	return (flags != "" && (flags == "up" || flags == "up|broadcast|multicast"))
+}
+
+// hasMatchingPods checks if there are any pods matching the EgressIP's namespace and pod selectors
+func (g *bridgeEIPAddrManager) hasMatchingPods(eip *egressipv1.EgressIP) (bool, error) {
+	namespaceSelector, err := metav1.LabelSelectorAsSelector(&eip.Spec.NamespaceSelector)
+	if err != nil {
+		return false, fmt.Errorf("failed to convert namespace selector: %v", err)
+	}
+
+	podSelector, err := metav1.LabelSelectorAsSelector(&eip.Spec.PodSelector)
+	if err != nil {
+		return false, fmt.Errorf("failed to convert pod selector: %v", err)
+	}
+
+	// Get all namespaces and filter by label selector
+	allNamespaces, err := g.namespaceLister.List(labels.Everything())
+	if err != nil {
+		return false, fmt.Errorf("failed to list all namespaces: %v", err)
+	}
+
+	for _, namespace := range allNamespaces {
+		namespaceLabels := labels.Set(namespace.Labels)
+		if !namespaceSelector.Matches(namespaceLabels) {
+			continue
+		}
+
+		// Get all pods in this namespace and filter by pod selector
+		allPods, err := g.podLister.Pods(namespace.Name).List(labels.Everything())
+		if err != nil {
+			return false, fmt.Errorf("failed to list pods in namespace %s: %v", namespace.Name, err)
+		}
+
+		for _, pod := range allPods {
+			podLabels := labels.Set(pod.Labels)
+			if !podSelector.Matches(podLabels) {
+				continue
+			}
+
+			// Check if pod is actually running and has IPs
+			if !util.PodCompleted(pod) && !util.PodWantsHostNetwork(pod) && len(pod.Status.PodIPs) > 0 {
+				return true, nil
+			}
+		}
+	}
+
+	return false, nil
+}
+
+func (g *bridgeEIPAddrManager) addEgressIP(eip *egressipv1.EgressIP) (bool, error) {
 	var isUpdated bool
 	if !util.IsEgressIPMarkSet(eip.Annotations) {
 		return isUpdated, nil
 	}
+
+	// First check if there are any matching pods for this EgressIP
+	hasMatchingPods, err := g.hasMatchingPods(eip)
+	if err != nil {
+		return isUpdated, fmt.Errorf("failed to check for matching pods: %v", err)
+	}
+	if !hasMatchingPods {
+		klog.V(5).Infof("EgressIP %s has no matching pods yet, skipping bridge IP assignment", eip.Name)
+		return isUpdated, nil
+	}
+
 	for _, status := range eip.Status.Items {
 		if status.Node != g.nodeName {
 			continue
@@ -223,6 +371,32 @@ func (g *BridgeEIPAddrManager) AddEgressIP(eip *egressipv1.EgressIP) (bool, erro
 		if err != nil {
 			return isUpdated, fmt.Errorf("failed to add EgressIP gateway config because unable to extract config from EgressIP obj: %v", err)
 		}
+
+		// Use longest prefix matching to determine the correct interface for this EgressIP
+		egressIP := net.ParseIP(status.EgressIP)
+		if egressIP == nil {
+			return isUpdated, fmt.Errorf("failed to parse EgressIP %s", status.EgressIP)
+		}
+
+		isEIPv4 := egressIP.To4() != nil
+		found, correctLink, err := g.findLinkOnSameNetworkAsIPUsingLPM(egressIP, isEIPv4, !isEIPv4)
+		if err != nil {
+			return isUpdated, fmt.Errorf("failed to find correct interface using LPM: %v", err)
+		}
+		if !found {
+			klog.Warningf("No suitable interface found for EgressIP %s using LPM", status.EgressIP)
+			return isUpdated, nil
+		}
+
+		// Only proceed if the bridge we're managing is the correct interface
+		if correctLink.Attrs().Name != g.bridgeName {
+			klog.V(5).Infof("EgressIP %s should be assigned to interface %s, not bridge %s, skipping",
+				status.EgressIP, correctLink.Attrs().Name, g.bridgeName)
+			return isUpdated, nil
+		}
+
+		klog.Infof("Adding EgressIP %s to bridge %s based on LPM and matching pods", status.EgressIP, g.bridgeName)
+
 		// must always add to cache before adding IP because we want to inform node ip handler that this is not a valid node IP
 		g.cache.insertMarkIP(pktMark, ip)
 		if err = g.addIPToAnnotation(ip); err != nil {
@@ -237,7 +411,7 @@ func (g *BridgeEIPAddrManager) AddEgressIP(eip *egressipv1.EgressIP) (bool, erro
 	return isUpdated, nil
 }
 
-func (g *BridgeEIPAddrManager) UpdateEgressIP(oldEIP, newEIP *egressipv1.EgressIP) (bool, error) {
+func (g *bridgeEIPAddrManager) UpdateEgressIP(oldEIP, newEIP *egressipv1.EgressIP) (bool, error) {
 	var isUpdated bool
 	// at most, one status item for this node will be found.
 	for _, oldStatus := range oldEIP.Status.Items {
@@ -293,7 +467,7 @@ func (g *BridgeEIPAddrManager) UpdateEgressIP(oldEIP, newEIP *egressipv1.EgressI
 	return isUpdated, nil
 }
 
-func (g *BridgeEIPAddrManager) DeleteEgressIP(eip *egressipv1.EgressIP) (bool, error) {
+func (g *bridgeEIPAddrManager) DeleteEgressIP(eip *egressipv1.EgressIP) (bool, error) {
 	var isUpdated bool
 	if !util.IsEgressIPMarkSet(eip.Annotations) {
 		return isUpdated, nil
@@ -322,7 +496,7 @@ func (g *BridgeEIPAddrManager) DeleteEgressIP(eip *egressipv1.EgressIP) (bool, e
 	return isUpdated, nil
 }
 
-func (g *BridgeEIPAddrManager) SyncEgressIP(objs []interface{}) error {
+func (g *bridgeEIPAddrManager) SyncEgressIP(objs []interface{}) error {
 	// caller must synchronise
 	annotIPs, err := g.getAnnotationIPs()
 	if err != nil {
@@ -380,7 +554,7 @@ func (g *BridgeEIPAddrManager) SyncEgressIP(objs []interface{}) error {
 
 // addIPToAnnotation adds an address to the collection of existing addresses stored in the nodes annotation. Caller
 // may repeat addition of addresses without care for duplicate addresses being added.
-func (g *BridgeEIPAddrManager) addIPToAnnotation(candidateIP net.IP) error {
+func (g *bridgeEIPAddrManager) addIPToAnnotation(candidateIP net.IP) error {
 	g.nodeAnnotationMu.Lock()
 	defer g.nodeAnnotationMu.Unlock()
 	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
@@ -412,7 +586,7 @@ func (g *BridgeEIPAddrManager) addIPToAnnotation(candidateIP net.IP) error {
 
 // deleteIPsFromAnnotation deletes address from annotation. If multiple users, callers must synchronise.
 // deletion of address that doesn't exist will not cause an error.
-func (g *BridgeEIPAddrManager) deleteIPsFromAnnotation(candidateIPs ...net.IP) error {
+func (g *bridgeEIPAddrManager) deleteIPsFromAnnotation(candidateIPs ...net.IP) error {
 	g.nodeAnnotationMu.Lock()
 	defer g.nodeAnnotationMu.Unlock()
 	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
@@ -446,15 +620,15 @@ func (g *BridgeEIPAddrManager) deleteIPsFromAnnotation(candidateIPs ...net.IP) e
 	})
 }
 
-func (g *BridgeEIPAddrManager) addIPBridge(ip net.IP) error {
+func (g *bridgeEIPAddrManager) addIPBridge(ip net.IP) error {
 	link, err := util.GetNetLinkOps().LinkByName(g.bridgeName)
 	if err != nil {
 		return fmt.Errorf("failed to get link obj by name %s: %v", g.bridgeName, err)
 	}
 	return g.addrManager.AddAddress(getEIPBridgeNetlinkAddress(ip, link.Attrs().Index))
 }
 
-func (g *BridgeEIPAddrManager) deleteIPBridge(ip net.IP) error {
+func (g *bridgeEIPAddrManager) deleteIPBridge(ip net.IP) error {
 	link, err := util.GetNetLinkOps().LinkByName(g.bridgeName)
 	if err != nil {
 		return fmt.Errorf("failed to get link obj by name %s: %v", g.bridgeName, err)
@@ -464,7 +638,7 @@ func (g *BridgeEIPAddrManager) deleteIPBridge(ip net.IP) error {
 
 // getAnnotationIPs retrieves the egress IP annotation from the current node Nodes object. If multiple users, callers must synchronise.
 // if annotation isn't present, empty set is returned
-func (g *BridgeEIPAddrManager) getAnnotationIPs() ([]net.IP, error) {
+func (g *bridgeEIPAddrManager) getAnnotationIPs() ([]net.IP, error) {
 	node, err := g.nodeLister.Get(g.nodeName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get node %s from lister: %v", g.nodeName, err)