Merge pull request #98 from fosrl/dev

1.4.2-s.1

internal/metrics/metrics.go

@@ -58,6 +58,7 @@ var (
 	udpPacketSizeBytes         observability.Histogram
 	holePunchEventsTotal       observability.Counter
 	proxyMappingActive         observability.UpDownCounter
+	relayUDPConnectionsActive  observability.UpDownCounter
 	sessionRebuiltTotal        observability.Counter
 	commPatternActive          observability.UpDownCounter
 	proxyCleanupRemovedTotal   observability.Counter
@@ -342,6 +343,11 @@ func createInstruments() error {
 	if err != nil {
 		return err
 	}
+	relayUDPConnectionsActive, err = newUpDownCounter("gerbil_relay_udp_connections_active",
+		"Number of open per-peer outbound UDP sockets held by the relay connection pool", "ifname")
+	if err != nil {
+		return err
+	}
 	sessionRebuiltTotal, err = newCounter("gerbil_session_rebuilt_total",
 		"Count of sessions rebuilt from communication patterns", "ifname")
 	if err != nil {
@@ -711,6 +717,13 @@ func RecordSession(ifname string, delta int64) {
 	activeSessions.Add(context.Background(), delta, observability.Labels{"ifname": ifname})
 }
 
+func RecordUDPConnection(ifname string, delta int64) {
+	if relayUDPConnectionsActive == nil {
+		return
+	}
+	relayUDPConnectionsActive.Add(context.Background(), delta, observability.Labels{"ifname": ifname})
+}
+
 func RecordSessionRebuilt(ifname string) {
 	if sessionRebuiltTotal == nil {
 		return

main.go

@@ -46,6 +46,7 @@ var (
 	doTrafficShaping bool
 	bandwidthLimit   string
 	ifbName          string // IFB device name for ingress traffic shaping
+	disableFirewall  bool
 )
 
 type WgConfig struct {
@@ -196,6 +197,7 @@ func main() {
 	proxyProtocolStr := os.Getenv("PROXY_PROTOCOL")
 	doTrafficShapingStr := os.Getenv("DO_TRAFFIC_SHAPING")
 	bandwidthLimitStr := os.Getenv("BANDWIDTH_LIMIT")
+	disableFirewallStr := os.Getenv("DISABLE_FIREWALL")
 
 	// Read metrics env vars (defaults applied by DefaultMetricsConfig; these override defaults).
 	metricsEnabled = true // default
@@ -316,6 +318,13 @@ func main() {
 		flag.BoolVar(&doTrafficShaping, "do-traffic-shaping", false, "Whether to set up traffic shaping rules for peers (requires tc command and root privileges)")
 	}
 
+	if disableFirewallStr != "" {
+		disableFirewall = strings.ToLower(disableFirewallStr) == "true"
+	}
+	if disableFirewallStr == "" {
+		flag.BoolVar(&disableFirewall, "disable-firewall", false, "Disable WireGuard firewall rules to allow all inbound traffic on the interface")
+	}
+
 	if bandwidthLimitStr != "" {
 		bandwidthLimit = bandwidthLimitStr
 	}
@@ -732,7 +741,9 @@ func ensureWireguardInterface(wgconfig WgConfig) error {
 		logger.Warn("Failed to ensure MSS clamping: %v", err)
 	}
 
-	if err := ensureWireguardFirewall(); err != nil {
+	if disableFirewall {
+		logger.Warn("Firewall disabled: all inbound traffic on %s will be allowed", interfaceName)
+	} else if err := ensureWireguardFirewall(); err != nil {
 		logger.Warn("Failed to ensure WireGuard firewall rules: %v", err)
 	}
 
@@ -1493,6 +1504,24 @@ func calculatePeerBandwidth() ([]PeerBandwidth, error) {
 	return peerBandwidths, nil
 }
 
+// defaultBandwidthReportBatchSize caps how many peer bandwidth readings are
+// sent in a single POST. Reporting every peer in one request grows unbounded
+// with fleet size and can exceed the remote server's request body limit,
+// which surfaces as "413 Payload Too Large" and silently drops that whole
+// report cycle. Overridable via GERBIL_BANDWIDTH_BATCH_SIZE.
+const defaultBandwidthReportBatchSize = 250
+
+var bandwidthReportBatchSize = loadBandwidthReportBatchSize()
+
+func loadBandwidthReportBatchSize() int {
+	if v := os.Getenv("GERBIL_BANDWIDTH_BATCH_SIZE"); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultBandwidthReportBatchSize
+}
+
 func reportPeerBandwidth(apiURL string) error {
 	bandwidths, err := calculatePeerBandwidth()
 	if err != nil {
@@ -1501,26 +1530,47 @@ func reportPeerBandwidth(apiURL string) error {
 		return fmt.Errorf("failed to calculate peer bandwidth: %v", err)
 	}
 
-	jsonData, err := json.Marshal(bandwidths)
+	if len(bandwidths) == 0 {
+		return nil
+	}
+
+	var batchErrs []error
+	for start := 0; start < len(bandwidths); start += bandwidthReportBatchSize {
+		end := start + bandwidthReportBatchSize
+		if end > len(bandwidths) {
+			end = len(bandwidths)
+		}
+
+		if err := sendPeerBandwidthBatch(apiURL, bandwidths[start:end]); err != nil {
+			metrics.RecordBandwidthReport("error")
+			batchErrs = append(batchErrs, err)
+			continue
+		}
+		metrics.RecordBandwidthReport("success")
+	}
+
+	if len(batchErrs) > 0 {
+		return fmt.Errorf("failed to report %d bandwidth batch(es): %w", len(batchErrs), errors.Join(batchErrs...))
+	}
+	return nil
+}
+
+func sendPeerBandwidthBatch(apiURL string, batch []PeerBandwidth) error {
+	jsonData, err := json.Marshal(batch)
 	if err != nil {
-		metrics.RecordBandwidthReport("error")
 		return fmt.Errorf("failed to marshal bandwidth data: %v", err)
 	}
 
 	resp, err := http.Post(apiURL, "application/json", bytes.NewBuffer(jsonData))
 	if err != nil {
-		metrics.RecordBandwidthReport("error")
 		return fmt.Errorf("failed to send bandwidth data: %v", err)
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		metrics.RecordBandwidthReport("error")
 		return fmt.Errorf("API returned non-OK status: %s", resp.Status)
 	}
 
-	// Record successful bandwidth report
-	metrics.RecordBandwidthReport("success")
 	return nil
 }
 

relay/relay.go

@@ -10,8 +10,11 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"os"
 	"runtime"
+	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/fosrl/gerbil/internal/metrics"
@@ -60,8 +63,31 @@ type PeerDestination struct {
 }
 
 type DestinationConn struct {
-	conn     *net.UDPConn
-	lastUsed time.Time
+	conn *net.UDPConn
+	// lastUsed is unix nanoseconds, read/written via atomic ops since it's
+	// touched from packet workers and the response goroutine concurrently
+	// with no lock, and is also scanned for LRU eviction below.
+	lastUsed atomic.Int64
+}
+
+// defaultMaxUDPConnections caps the number of concurrent per-peer outbound
+// UDP sockets the relay will keep open in s.connections. Without a cap, a
+// burst of peer churn creates sockets faster than the 5-minute idle cleanup
+// can reap them, exhausting the host's ephemeral port range or fd ulimit.
+// That surfaces as "dial udp ...: resource temporarily unavailable" on every
+// subsequent packet and pegs the CPU logging the flood (outage 2026-07-03,
+// recurrence 2026-07-05). Overridable via GERBIL_MAX_UDP_CONNECTIONS.
+const defaultMaxUDPConnections = 8192
+
+var maxUDPConnections = loadMaxUDPConnections()
+
+func loadMaxUDPConnections() int64 {
+	if v := os.Getenv("GERBIL_MAX_UDP_CONNECTIONS"); v != "" {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n > 0 {
+			return n
+		}
+	}
+	return defaultMaxUDPConnections
 }
 
 // Type for storing WireGuard handshake information
@@ -172,10 +198,13 @@ type UDPProxyServer struct {
 	conn          *net.UDPConn
 	proxyMappings sync.Map // map[string]ProxyMapping where key is "ip:port"
 	connections   sync.Map // map[string]*DestinationConn where key is destination "ip:port"
-	privateKey    wgtypes.Key
-	packetChan    chan Packet
-	ctx           context.Context
-	cancel        context.CancelFunc
+	// connectionCount mirrors len(connections) without an O(n) sync.Map walk,
+	// so the cap check in getOrCreateConnection is cheap on the hot path.
+	connectionCount atomic.Int64
+	privateKey      wgtypes.Key
+	packetChan      chan Packet
+	ctx             context.Context
+	cancel          context.CancelFunc
 
 	// Session tracking for WireGuard peers
 	// Key format: "senderIndex:receiverIndex"
@@ -862,10 +891,17 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr
 	// Check if we have an existing connection
 	if conn, ok := s.connections.Load(key); ok {
 		destConn := conn.(*DestinationConn)
-		destConn.lastUsed = time.Now()
+		destConn.lastUsed.Store(time.Now().UnixNano())
 		return destConn.conn, nil
 	}
 
+	// Enforce a hard cap on concurrent sockets so a burst of peer churn can't
+	// exhaust the host's ephemeral ports/fds. Evict the least-recently-used
+	// connection to make room instead of growing unbounded.
+	if s.connectionCount.Load() >= maxUDPConnections {
+		s.evictLRUConnection()
+	}
+
 	// Create new connection
 	newConn, err := net.DialUDP("udp", nil, destAddr)
 	if err != nil {
@@ -873,11 +909,17 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr
 		return nil, fmt.Errorf("failed to create UDP connection: %v", err)
 	}
 
-	// Store the new connection
-	s.connections.Store(key, &DestinationConn{
-		conn:     newConn,
-		lastUsed: time.Now(),
-	})
+	destConn := &DestinationConn{conn: newConn}
+	destConn.lastUsed.Store(time.Now().UnixNano())
+
+	// Store the new connection. If another goroutine raced us and already
+	// created one for this key, close ours and use theirs instead.
+	if existing, loaded := s.connections.LoadOrStore(key, destConn); loaded {
+		newConn.Close()
+		return existing.(*DestinationConn).conn, nil
+	}
+	s.connectionCount.Add(1)
+	metrics.RecordUDPConnection(relayIfname, 1)
 
 	// Start a goroutine to handle responses
 	go s.handleResponses(newConn, destAddr, remoteAddr)
@@ -885,6 +927,33 @@ func (s *UDPProxyServer) getOrCreateConnection(destAddr *net.UDPAddr, remoteAddr
 	return newConn, nil
 }
 
+// evictLRUConnection closes and removes the least-recently-used destination
+// connection so a new one can be created under the concurrent connection cap.
+func (s *UDPProxyServer) evictLRUConnection() {
+	var oldestKey interface{}
+	var oldestConn *DestinationConn
+	var oldestTime int64
+
+	s.connections.Range(func(key, value interface{}) bool {
+		destConn := value.(*DestinationConn)
+		lu := destConn.lastUsed.Load()
+		if oldestKey == nil || lu < oldestTime {
+			oldestKey = key
+			oldestConn = destConn
+			oldestTime = lu
+		}
+		return true
+	})
+
+	if oldestKey != nil {
+		s.connections.Delete(oldestKey)
+		oldestConn.conn.Close()
+		s.connectionCount.Add(-1)
+		metrics.RecordUDPConnection(relayIfname, -1)
+		metrics.RecordProxyCleanupRemoved(relayIfname, "conn_evicted", 1)
+	}
+}
+
 func (s *UDPProxyServer) handleResponses(conn *net.UDPConn, destAddr *net.UDPAddr, remoteAddr *net.UDPAddr) {
 	buffer := make([]byte, 1500)
 	for {
@@ -936,22 +1005,32 @@ func (s *UDPProxyServer) handleResponses(conn *net.UDPConn, destAddr *net.UDPAdd
 
 // Add a cleanup method to periodically remove idle connections
 func (s *UDPProxyServer) cleanupIdleConnections() {
-	ticker := time.NewTicker(5 * time.Minute)
+	// Ticker interval and idle threshold were previously 5min/10min, meaning
+	// a socket could sit open for up to 15 minutes after going idle. Under a
+	// reconnect/churn burst that lag is enough to exhaust ephemeral ports
+	// before cleanup catches up, so both are tightened here.
+	ticker := time.NewTicker(1 * time.Minute)
 	defer ticker.Stop()
 	for {
 		select {
 		case <-ticker.C:
 			cleanupStart := time.Now()
-			now := time.Now()
+			now := time.Now().UnixNano()
+			removed := int64(0)
 			s.connections.Range(func(key, value interface{}) bool {
 				destConn := value.(*DestinationConn)
-				if now.Sub(destConn.lastUsed) > 10*time.Minute {
+				if now-destConn.lastUsed.Load() > int64(5*time.Minute) {
 					destConn.conn.Close()
 					s.connections.Delete(key)
-					metrics.RecordProxyCleanupRemoved(relayIfname, "conn", 1)
+					removed++
 				}
 				return true
 			})
+			if removed > 0 {
+				s.connectionCount.Add(-removed)
+				metrics.RecordUDPConnection(relayIfname, -removed)
+				metrics.RecordProxyCleanupRemoved(relayIfname, "conn", removed)
+			}
 			metrics.RecordProxyIdleCleanupDuration(relayIfname, "conn", time.Since(cleanupStart).Seconds())
 		case <-s.ctx.Done():
 			return
@@ -1113,6 +1192,10 @@ func (s *UDPProxyServer) clearConnectionsForWGIP(wgIP string) {
 	for _, key := range keysToDelete {
 		s.connections.Delete(key)
 	}
+	if len(keysToDelete) > 0 {
+		s.connectionCount.Add(-int64(len(keysToDelete)))
+		metrics.RecordUDPConnection(relayIfname, -int64(len(keysToDelete)))
+	}
 
 	logger.Info("Cleared %d connections for WG IP: %s", len(keysToDelete), wgIP)
 }