Fixes for PR

Author: fdurand

go/cmd/pfudpproxy/config.go

@@ -19,6 +19,7 @@ const (
 
 	PortNetFlow = 2055
 	PortSFlow   = 6343
+	PortIPFIX   = 4739
 )
 
 // ProxyConfig holds the configuration for the UDP proxy
@@ -53,7 +54,7 @@ func getHealthCheckPort(ctx context.Context) int {
 // LoadConfig loads the proxy configuration from pfconfig
 func LoadConfig(ctx context.Context) (*ProxyConfig, error) {
 	config := &ProxyConfig{
-		Ports:               []int{PortNetFlow, PortSFlow},
+		Ports:               []int{PortNetFlow, PortSFlow, PortIPFIX},
 		HealthCheckPort:     getHealthCheckPort(ctx),
 		HealthCheckPath:     DefaultHealthCheckPath,
 		HealthCheckInterval: DefaultHealthCheckInterval,

go/cmd/pfudpproxy/healthcheck.go

@@ -78,12 +78,12 @@ func (hc *HealthChecker) checkAllBackends(ctx context.Context) {
 	log.LoggerWContext(ctx).Debug(fmt.Sprintf("Running health checks for %d backends", len(backends)))
 
 	var wg sync.WaitGroup
-	for _, backend := range backends {
+	for i := range backends {
 		wg.Add(1)
-		go func(b *Backend) {
+		go func(b Backend) {
 			defer wg.Done()
 			hc.checkBackend(ctx, b)
-		}(backend)
+		}(backends[i])
 	}
 	wg.Wait()
 
@@ -97,7 +97,9 @@ func (hc *HealthChecker) checkAllBackends(ctx context.Context) {
 }
 
 // checkBackend checks the health of a single backend.
-func (hc *HealthChecker) checkBackend(ctx context.Context, backend *Backend) {
+// backend is received by value (a snapshot from GetAllBackends) so reads
+// of its fields are safe without holding the load balancer lock.
+func (hc *HealthChecker) checkBackend(ctx context.Context, backend Backend) {
 	url := fmt.Sprintf("https://%s:%d%s",
 		backend.ManagementIP,
 		hc.config.HealthCheckPort,

go/cmd/pfudpproxy/loadbalancer.go

@@ -21,27 +21,33 @@ func NewLoadBalancer(backends []*Backend) *LoadBalancer {
 	}
 }
 
-// GetPrimary returns the first healthy backend (failover mode).
+// GetPrimary returns a snapshot of the first healthy backend (failover mode).
 // Returns nil if no healthy backend is available.
+// The returned value is a copy; mutating it does not affect the load balancer.
 func (lb *LoadBalancer) GetPrimary() *Backend {
 	lb.mu.RLock()
 	defer lb.mu.RUnlock()
 
 	for _, backend := range lb.backends {
 		if backend.Healthy {
-			return backend
+			copy := *backend
+			return &copy
 		}
 	}
 	return nil
 }
 
-// GetAllBackends returns a copy of all backends for health checking.
-func (lb *LoadBalancer) GetAllBackends() []*Backend {
+// GetAllBackends returns a snapshot of all backends for health checking.
+// Each element is a struct copy; callers can read fields safely without
+// the load balancer lock.
+func (lb *LoadBalancer) GetAllBackends() []Backend {
 	lb.mu.RLock()
 	defer lb.mu.RUnlock()
 
-	backends := make([]*Backend, len(lb.backends))
-	copy(backends, lb.backends)
+	backends := make([]Backend, len(lb.backends))
+	for i, b := range lb.backends {
+		backends[i] = *b
+	}
 	return backends
 }
 

go/cmd/pfudpproxy/proxy.go

@@ -26,14 +26,25 @@ type UDPProxy struct {
 	running   bool
 	stopChan  chan struct{}
 	wg        sync.WaitGroup
+
+	// fwdConn is a shared unbound UDP connection used by all forwarders.
+	// Because it is not bound to a specific destination, WriteToUDP can
+	// send to any backend without opening a new socket per packet.
+	fwdConn *net.UDPConn
+
+	// addrCache maps "ip:port" to a resolved *net.UDPAddr so we don't
+	// call ResolveUDPAddr on every packet.
+	addrCache   map[string]*net.UDPAddr
+	addrCacheMu sync.RWMutex
 }
 
 // NewUDPProxy creates a new UDP proxy.
 func NewUDPProxy(config *ProxyConfig, lb *LoadBalancer) *UDPProxy {
 	return &UDPProxy{
-		config:   config,
-		lb:       lb,
-		stopChan: make(chan struct{}),
+		config:    config,
+		lb:        lb,
+		stopChan:  make(chan struct{}),
+		addrCache: make(map[string]*net.UDPAddr),
 	}
 }
 
@@ -49,6 +60,14 @@ func (p *UDPProxy) Start(ctx context.Context) {
 
 	log.LoggerWContext(ctx).Info("Starting UDP proxy")
 
+	// Open a single unbound UDP socket for all outbound forwarding.
+	fwd, err := net.ListenUDP("udp", nil)
+	if err != nil {
+		log.LoggerWContext(ctx).Error(fmt.Sprintf("Failed to open forwarding socket: %s", err.Error()))
+		return
+	}
+	p.fwdConn = fwd
+
 	for _, port := range p.config.Ports {
 		p.wg.Add(1)
 		go func(port int) {
@@ -76,6 +95,11 @@ func (p *UDPProxy) Stop(ctx context.Context) {
 		}
 	}
 
+	// Close the shared forwarding socket
+	if p.fwdConn != nil {
+		p.fwdConn.Close()
+	}
+
 	// Wait for all goroutines to finish
 	done := make(chan struct{})
 	go func() {
@@ -106,6 +130,11 @@ func (p *UDPProxy) UpdateConfig(ctx context.Context, newConfig *ProxyConfig) {
 	}
 
 	p.config = newConfig
+
+	// Flush the address cache so stale entries don't survive a backend change.
+	p.addrCacheMu.Lock()
+	p.addrCache = make(map[string]*net.UDPAddr)
+	p.addrCacheMu.Unlock()
 }
 
 // listenAndForward listens on VIP:port and forwards packets to healthy backends.
@@ -168,6 +197,27 @@ func (p *UDPProxy) listenAndForward(ctx context.Context, port int) {
 	}
 }
 
+// resolveAddr returns a cached *net.UDPAddr for the given key (ip:port),
+// resolving and caching it on the first call.
+func (p *UDPProxy) resolveAddr(key string) (*net.UDPAddr, error) {
+	p.addrCacheMu.RLock()
+	addr, ok := p.addrCache[key]
+	p.addrCacheMu.RUnlock()
+	if ok {
+		return addr, nil
+	}
+
+	addr, err := net.ResolveUDPAddr("udp", key)
+	if err != nil {
+		return nil, err
+	}
+
+	p.addrCacheMu.Lock()
+	p.addrCache[key] = addr
+	p.addrCacheMu.Unlock()
+	return addr, nil
+}
+
 // forwardPacket forwards a UDP packet to the primary healthy backend.
 func (p *UDPProxy) forwardPacket(ctx context.Context, data []byte, srcAddr *net.UDPAddr, port int) {
 	backend := p.lb.GetPrimary()
@@ -176,32 +226,23 @@ func (p *UDPProxy) forwardPacket(ctx context.Context, data []byte, srcAddr *net.
 		return
 	}
 
-	// Create destination address using backend's management IP and same port
-	dstAddr := fmt.Sprintf("%s:%d", backend.ManagementIP, port)
-	udpDstAddr, err := net.ResolveUDPAddr("udp", dstAddr)
+	// Resolve (or retrieve from cache) the destination address
+	dstKey := fmt.Sprintf("%s:%d", backend.ManagementIP, port)
+	udpDstAddr, err := p.resolveAddr(dstKey)
 	if err != nil {
 		log.LoggerWContext(ctx).Error(fmt.Sprintf("Failed to resolve destination address %s: %s",
-			dstAddr, err.Error()))
-		return
-	}
-
-	// Create a new UDP connection for forwarding
-	// Using a new connection each time since NetFlow/sFlow are fire-and-forget
-	conn, err := net.DialUDP("udp", nil, udpDstAddr)
-	if err != nil {
-		log.LoggerWContext(ctx).Error(fmt.Sprintf("Failed to connect to backend %s: %s",
-			dstAddr, err.Error()))
+			dstKey, err.Error()))
 		return
 	}
-	defer conn.Close()
 
-	_, err = conn.Write(data)
+	// Use the shared unbound socket to send to the backend
+	_, err = p.fwdConn.WriteToUDP(data, udpDstAddr)
 	if err != nil {
 		log.LoggerWContext(ctx).Error(fmt.Sprintf("Failed to forward packet to %s: %s",
-			dstAddr, err.Error()))
+			dstKey, err.Error()))
 		return
 	}
 
 	log.LoggerWContext(ctx).Debug(fmt.Sprintf("Forwarded %d bytes from %s to %s",
-		len(data), srcAddr.String(), dstAddr))
+		len(data), srcAddr.String(), dstKey))
 }

lib/pf/iptables.pm

@@ -1563,7 +1563,7 @@ sub iptables_fingerbank_collector_rules {
 
 =item iptables_pfudpproxy_rules
 
-Iptable rules for pfudpproxy service (UDP proxy for NetFlow/sFlow in cluster mode)
+Iptable rules for pfudpproxy service (UDP proxy for NetFlow/sFlow/IPFIX in cluster mode)
 
 =cut
 
@@ -1582,6 +1582,8 @@ sub iptables_pfudpproxy_rules {
         util_safe_push( "-i $tint -p udp -m udp --dport 2055 --jump ACCEPT", $chains->{'filter'}{'INPUT'} );
         # Port 6343 - sFlow (UDP)
         util_safe_push( "-i $tint -p udp -m udp --dport 6343 --jump ACCEPT", $chains->{'filter'}{'INPUT'} );
+        # Port 4739 - IPFIX (UDP)
+        util_safe_push( "-i $tint -p udp -m udp --dport 4739 --jump ACCEPT", $chains->{'filter'}{'INPUT'} );
         # Convert to JSON and save to file
         util_create_service_rules($chains);
     } else {